mz/video-gen
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

超分功能完成

Browse Source
This commit is contained in:
GinHa
2026-07-16 15:02:03 +08:00
parent 51673453d8
commit 3db586cd60
49 changed files with 4423 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-admin/src/App.tsx
+2
View File
@@ -32,6 +32,7 @@ import AdminShotReplications from './pages/AdminShotReplications';
import AdminShotTaskSetDetail from './pages/AdminShotTaskSetDetail';
import AdminReplicationProjectDetail from './pages/AdminReplicationProjectDetail';
import AdminVideoPromptSchemaConfig from './pages/AdminVideoPromptSchemaConfig';
import AdminVideoUpscale from './pages/AdminVideoUpscale';
import AdminContactRequests from './pages/AdminContactRequests';
import AdminHomeMaterials from './pages/AdminHomeMaterials';
import AdminPreTestTemplates from './pages/AdminPreTestTemplates';
@@ -98,6 +99,7 @@ const App = () => {
<Route path="payment-stats" element={<AdminPaymentStats />} />
<Route path="settings" element={<AdminSettings />} />
<Route path="video-prompt-schema-config" element={<AdminVideoPromptSchemaConfig />} />
<Route path="video-upscale" element={<AdminVideoUpscale />} />
<Route path="notifications" element={<AdminNotificationManager />} />
<Route path="oauthapp-list" element={<AdminOauthAppList />} />
<Route path="operation-logs" element={<AdminOperationLogs />} />
video-gen-admin/src/api/index.ts
+13
View File
@@ -17,6 +17,7 @@ import type {
AdminTeam, AdminTeamListResponse, AdminTeamOption, AdminTeamPayload, AdminTeamQueryParams,
PrivatePortraitConfig, PrivatePortraitProjectListOut, PrivatePortraitAssetListOut,
AdminUploadFileResult, AdminUploadResourceType, AdminUploadScene,
VideoUpscaleConfigOut, VideoUpscaleConfigSavePayload,
} from '../types';
import type {
@@ -41,6 +42,18 @@ import type {
HomeMaterialWatermarkQueryParams,
} from '../types';
// ── Video Upscale ────────────────────────────────────────
export async function getVideoUpscaleConfig(): Promise<VideoUpscaleConfigOut> {
return api.get<VideoUpscaleConfigOut>('/admin/video-upscale/config');
}
export async function saveVideoUpscaleConfig(payload: VideoUpscaleConfigSavePayload): Promise<VideoUpscaleConfigOut> {
return api.put<VideoUpscaleConfigOut>('/admin/video-upscale/config', payload);
}
// ── Auth ──────────────────────────────────────────────────
export async function login(username: string, password: string, captchaToken?: string, rememberMe?: boolean): Promise<User> {
video-gen-admin/src/pages/AdminVideoUpscale.tsx
+262
View File
@@ -0,0 +1,262 @@
import React, { useCallback, useEffect, useState } from 'react';
import {
App,
Button,
Card,
Col,
Empty,
Row,
Select,
Space,
Spin,
Switch,
Tag,
Typography,
} from 'antd';
import { DeleteOutlined, PlusOutlined, ReloadOutlined, SaveOutlined } from '@ant-design/icons';
import { getVideoUpscaleConfig, saveVideoUpscaleConfig } from '../api';
import type {
VideoUpscaleConfigData,
VideoUpscaleProcessorKey,
VideoUpscaleResolutionRule,
} from '../types';
const { Title, Text, Paragraph } = Typography;
const PROCESSORS: Array<{ key: VideoUpscaleProcessorKey; label: string }> = [
{ key: 'local_ffmpeg_crop_v1', label: '本地 FFmpeg（crop）' },
{ key: 'volc_standard_v1', label: '火山画质增强（标准版）' },
{ key: 'volc_professional_v1', label: '火山画质增强（专业版）' },
{ key: 'volc_large_model_v1', label: '火山画质增强（大模型）' },
];
const RESOLUTION_OPTIONS = ['480p', '720p', '1080p', '2K', '4K'].map((value) => ({
label: value,
value,
}));
const defaultRule = (): VideoUpscaleResolutionRule => ({
targetResolution: '1080p',
providerGenerationResolution: '720p',
processorKey: 'local_ffmpeg_crop_v1',
enabled: true,
});
function normalizeConfig(data: VideoUpscaleConfigData): VideoUpscaleConfigData {
return {
enabled: !!data.enabled,
version: Number(data.version || 1),
deleteSourceAfterSuccess: data.deleteSourceAfterSuccess !== false,
rules: Array.isArray(data.rules) ? data.rules.map((rule) => ({ ...rule })) : [],
};
}
function toSavePayload(data: VideoUpscaleConfigData) {
return {
data: {
enabled: data.enabled,
version: data.version,
delete_source_after_success: data.deleteSourceAfterSuccess,
rules: data.rules.map((rule) => ({
target_resolution: rule.targetResolution,
provider_generation_resolution: rule.providerGenerationResolution,
processor_key: rule.processorKey,
enabled: rule.enabled,
})),
},
};
}
const AdminVideoUpscale: React.FC = () => {
const { message, modal } = App.useApp();
const [loading, setLoading] = useState(true);
const [saving, setSaving] = useState(false);
const [config, setConfig] = useState<VideoUpscaleConfigData | null>(null);
const load = useCallback(async () => {
setLoading(true);
try {
const result = await getVideoUpscaleConfig();
setConfig(normalizeConfig(result.data));
} catch (error) {
message.error(error instanceof Error ? error.message : '读取视频超分配置失败');
} finally {
setLoading(false);
}
}, [message]);
useEffect(() => {
void load();
}, [load]);
const updateRule = (index: number, patch: Partial<VideoUpscaleResolutionRule>) => {
setConfig((current) => {
if (!current) return current;
return {
...current,
rules: current.rules.map((item, itemIndex) => itemIndex === index ? { ...item, ...patch } : item),
};
});
};
const removeRule = (index: number) => {
setConfig((current) => current ? {
...current,
rules: current.rules.filter((_, itemIndex) => itemIndex !== index),
} : current);
};
const addRule = () => {
setConfig((current) => current ? { ...current, rules: [...current.rules, defaultRule()] } : current);
};
const save = async () => {
if (!config) return;
const targets = config.rules
.filter((item) => item.enabled)
.map((item) => item.targetResolution.trim().toLowerCase());
if (new Set(targets).size !== targets.length) {
message.error('同一个客户目标分辨率只能存在一条启用规则');
return;
}
setSaving(true);
try {
const result = await saveVideoUpscaleConfig(toSavePayload(config));
setConfig(normalizeConfig(result.data));
message.success(`视频超分配置已保存，版本 ${result.data.version}`);
} catch (error) {
message.error(error instanceof Error ? error.message : '保存失败');
} finally {
setSaving(false);
}
};
if (loading || !config) {
return (
<div style={{ minHeight: 360, display: 'flex', alignItems: 'center', justifyContent: 'center' }}>
<Spin size="large" />
</div>
);
}
return (
<Space direction="vertical" size={16} style={{ width: '100%' }}>
<Card>
<Row justify="space-between" align="middle" gutter={[16, 16]}>
<Col>
<Title level={3} style={{ margin: 0 }}>视频超分配置</Title>
<Paragraph type="secondary" style={{ margin: '8px 0 0' }}>
规则只按客户选择的目标分辨率匹配。视频比例和最终像素由客户任务参数在创建任务时自动计算并固化快照。
</Paragraph>
</Col>
<Col>
<Space>
<Tag>配置版本 {config.version}</Tag>
<Button icon={<ReloadOutlined />} onClick={() => void load()}>重新加载</Button>
<Button type="primary" icon={<SaveOutlined />} loading={saving} onClick={() => void save()}>
保存配置
</Button>
</Space>
</Col>
</Row>
<Space direction="vertical" size={14} style={{ marginTop: 20 }}>
<Space>
<Text strong>全局开启超分</Text>
<Switch
checked={config.enabled}
onChange={(enabled) => setConfig({ ...config, enabled })}
/>
<Text type="secondary">关闭后新任务走原流程，已经创建的任务仍按自身快照执行。</Text>
</Space>
<Space>
<Text strong>超分成功后删除源视频</Text>
<Switch
checked={config.deleteSourceAfterSuccess}
onChange={(deleteSourceAfterSuccess) => setConfig({ ...config, deleteSourceAfterSuccess })}
/>
<Text type="secondary">
默认开启。关闭仅用于调试，会保留超分前源视频并持续占用服务器磁盘；超分失败时始终保留源视频。
</Text>
</Space>
</Space>
</Card>
<Card
title="目标分辨率规则"
extra={<Button icon={<PlusOutlined />} onClick={addRule}>新增规则</Button>}
>
{config.rules.length === 0 ? (
<Empty description="暂无规则；未匹配规则的视频任务继续走原流程" />
) : (
<Space direction="vertical" size={12} style={{ width: '100%' }}>
{config.rules.map((rule, index) => {
const duplicate = rule.enabled && config.rules.filter(
(item) => item.enabled && item.targetResolution.toLowerCase() === rule.targetResolution.toLowerCase(),
).length > 1;
return (
<Card key={`${index}-${rule.targetResolution}`} size="small">
<Row gutter={[12, 12]} align="bottom">
<Col xs={24} md={5}>
<Text type="secondary">客户选择分辨率</Text>
<Select
value={rule.targetResolution}
options={RESOLUTION_OPTIONS}
style={{ width: '100%', marginTop: 4 }}
onChange={(value) => updateRule(index, { targetResolution: value })}
status={duplicate ? 'error' : undefined}
/>
</Col>
<Col xs={24} md={5}>
<Text type="secondary">实际生成分辨率</Text>
<Select
value={rule.providerGenerationResolution}
options={RESOLUTION_OPTIONS}
style={{ width: '100%', marginTop: 4 }}
onChange={(value) => updateRule(index, { providerGenerationResolution: value })}
/>
</Col>
<Col xs={24} md={7}>
<Text type="secondary">处理方式</Text>
<Select
value={rule.processorKey}
options={PROCESSORS.map((item) => ({ value: item.key, label: item.label }))}
style={{ width: '100%', marginTop: 4 }}
onChange={(value) => updateRule(index, { processorKey: value })}
/>
</Col>
<Col xs={12} md={2}>
<Text type="secondary">启用</Text>
<div style={{ marginTop: 8 }}>
<Switch checked={rule.enabled} onChange={(enabled) => updateRule(index, { enabled })} />
</div>
</Col>
<Col xs={12} md={2}>
<Button
danger
icon={<DeleteOutlined />}
onClick={() => modal.confirm({
title: '删除这条超分规则？',
onOk: () => removeRule(index),
})}
/>
</Col>
<Col span={24}>
<Text type="secondary">
客户选择该目标分辨率时，无论横屏、竖屏或方形比例，都按本规则选择的实际分辨率生成；最终像素由任务比例自动计算。
</Text>
</Col>
</Row>
</Card>
);
})}
</Space>
)}
</Card>
</Space>
);
};
export default AdminVideoUpscale;
video-gen-admin/src/types/index.ts
+45
View File
@@ -1308,3 +1308,48 @@ export interface PrivatePortraitSelectableAssetListOut {
page: number;
pageSize: number;
}
// ── Video Upscale ────────────────────────────────────────
export type VideoUpscaleProcessorKey =
| 'local_ffmpeg_crop_v1'
| 'volc_large_model_v1'
| 'volc_standard_v1'
| 'volc_professional_v1';
export interface VideoUpscaleResolutionRule {
targetResolution: '480p' | '720p' | '1080p' | '2K' | '4K';
providerGenerationResolution: '480p' | '720p' | '1080p' | '2K' | '4K';
processorKey: VideoUpscaleProcessorKey;
enabled: boolean;
}
export interface VideoUpscaleConfigData {
enabled: boolean;
version: number;
deleteSourceAfterSuccess: boolean;
rules: VideoUpscaleResolutionRule[];
}
export interface VideoUpscaleConfigOut {
id?: string | null;
key: string;
description?: string | null;
data: VideoUpscaleConfigData;
createdAt?: string | null;
updatedAt?: string | null;
}
export interface VideoUpscaleConfigSavePayload {
data: {
enabled: boolean;
version: number;
delete_source_after_success: boolean;
rules: Array<{
target_resolution: string;
provider_generation_resolution: string;
processor_key: VideoUpscaleProcessorKey;
enabled: boolean;
}>;
};
}
video-gen-api/.env
+1
View File
@@ -58,6 +58,7 @@ VIDEO_COVER_TIMEOUT_SECONDS=15
VIDEO_COVER_FORMAT=png
# VOLC
VOLC_API_KEY=AKLTOWMwMjVhNzg0OGE2NDMwZWJkYWIyNzM3ZmMxMjc5NTQ
VOLC_ACCESS_KEY_ID=AKLTYWY5Yjc5YjM3N2IwNDc3M2I3NTU2YjlmNTczYzQzMmM
VOLC_SECRET_ACCESS_KEY=TXpjM01HUTFZMlV5TUdKbE5Ea3lNRGhqTUdSak16UTFOV0ptTW1SaE5XRQ==
video-gen-api/alembic/versions/3f47680a71d0_add_video_upscale_pipeline.py
+180
View File
@@ -0,0 +1,180 @@
"""add video upscale pipeline
Revision ID: 3f47680a71d0
Revises: abae3e1c70f7
Create Date: 2026-07-16 11:08:16.449960
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "3f47680a71d0"
down_revision: Union[str, None] = "abae3e1c70f7"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.add_column(
"chat_generation_tasks",
sa.Column("provider_generation_resolution", sa.String(length=16), nullable=True),
)
op.add_column(
"chat_generation_tasks",
sa.Column(
"video_upscale_enabled_snapshot",
sa.Boolean(),
server_default=sa.text("false"),
nullable=False,
),
)
op.add_column(
"chat_generation_tasks",
sa.Column("video_upscale_snapshot_json", sa.Text(), nullable=True),
)
op.add_column(
"generation_records",
sa.Column("provider_generation_resolution", sa.String(length=16), nullable=True),
)
op.add_column(
"generation_records",
sa.Column(
"video_upscale_enabled_snapshot",
sa.Boolean(),
server_default=sa.text("false"),
nullable=False,
),
)
op.add_column(
"generation_records",
sa.Column("video_upscale_snapshot_json", sa.Text(), nullable=True),
)
op.add_column(
"generation_records",
sa.Column("pipeline_stage", sa.String(length=48), nullable=True),
)
op.create_index(
"ix_generation_records_pipeline_stage",
"generation_records",
["pipeline_stage"],
unique=False,
)
op.create_table(
"video_upscale_tasks",
sa.Column("id", sa.String(length=32), nullable=False),
sa.Column("chat_generation_task_id", sa.String(length=32), nullable=True),
sa.Column("generation_record_id", sa.String(length=32), nullable=True),
sa.Column("status", sa.String(length=32), server_default="pending", nullable=False),
sa.Column("stage", sa.String(length=48), server_default="upscale_queued", nullable=False),
sa.Column("processor_key", sa.String(length=64), nullable=False),
sa.Column("attempt_count", sa.Integer(), server_default="0", nullable=False),
sa.Column("failure_count", sa.Integer(), server_default="0", nullable=False),
sa.Column("manual_retry_count", sa.Integer(), server_default="0", nullable=False),
sa.Column("next_retry_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("last_error", sa.Text(), nullable=True),
sa.Column("source_local_path", sa.Text(), nullable=True),
sa.Column("source_file_size_bytes", sa.BigInteger(), server_default="0", nullable=False),
sa.Column("source_width", sa.Integer(), nullable=True),
sa.Column("source_height", sa.Integer(), nullable=True),
sa.Column("source_duration_seconds", sa.Float(), nullable=True),
sa.Column("source_deleted_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("source_delete_error", sa.Text(), nullable=True),
sa.Column("source_remote_url", sa.Text(), nullable=True),
sa.Column("source_remote_url_signed_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("source_remote_url_expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("source_remote_url_last_probe_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("source_remote_url_probe_status", sa.String(length=32), nullable=True),
sa.Column("input_source_type", sa.String(length=32), nullable=True),
sa.Column("input_source_fallback_count", sa.Integer(), server_default="0", nullable=False),
sa.Column("target_width", sa.Integer(), nullable=False),
sa.Column("target_height", sa.Integer(), nullable=False),
sa.Column("effective_target_width", sa.Integer(), nullable=True),
sa.Column("effective_target_height", sa.Integer(), nullable=True),
sa.Column("provider_task_id", sa.String(length=160), nullable=True),
sa.Column("provider_request_json", sa.Text(), nullable=True),
sa.Column("provider_response_json", sa.Text(), nullable=True),
sa.Column("provider_output_url", sa.Text(), nullable=True),
sa.Column("provider_output_url_expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("provider_submitted_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("final_local_path", sa.Text(), nullable=True),
sa.Column("final_resource_url", sa.Text(), nullable=True),
sa.Column("final_file_size_bytes", sa.BigInteger(), server_default="0", nullable=False),
sa.Column("celery_task_id", sa.String(length=160), nullable=True),
sa.Column("lease_token", sa.String(length=64), nullable=True),
sa.Column("lease_until", sa.DateTime(timezone=True), nullable=True),
sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("failed_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.CheckConstraint(
"(chat_generation_task_id IS NOT NULL AND generation_record_id IS NULL) OR "
"(chat_generation_task_id IS NULL AND generation_record_id IS NOT NULL)",
name="ck_video_upscale_tasks_exactly_one_owner",
),
sa.ForeignKeyConstraint(
["chat_generation_task_id"],
["chat_generation_tasks.id"],
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["generation_record_id"],
["generation_records.id"],
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(
"idx_video_upscale_tasks_provider_task_id",
"video_upscale_tasks",
["provider_task_id"],
unique=False,
)
op.create_index(
"idx_video_upscale_tasks_status_lease",
"video_upscale_tasks",
["status", "lease_until"],
unique=False,
)
op.create_index(
"idx_video_upscale_tasks_status_next_retry",
"video_upscale_tasks",
["status", "next_retry_at"],
unique=False,
)
op.create_index(
"uq_video_upscale_tasks_chat_task",
"video_upscale_tasks",
["chat_generation_task_id"],
unique=True,
)
op.create_index(
"uq_video_upscale_tasks_generation_record",
"video_upscale_tasks",
["generation_record_id"],
unique=True,
)
def downgrade() -> None:
op.drop_index("uq_video_upscale_tasks_generation_record", table_name="video_upscale_tasks")
op.drop_index("uq_video_upscale_tasks_chat_task", table_name="video_upscale_tasks")
op.drop_index("idx_video_upscale_tasks_status_next_retry", table_name="video_upscale_tasks")
op.drop_index("idx_video_upscale_tasks_status_lease", table_name="video_upscale_tasks")
op.drop_index("idx_video_upscale_tasks_provider_task_id", table_name="video_upscale_tasks")
op.drop_table("video_upscale_tasks")
op.drop_index("ix_generation_records_pipeline_stage", table_name="generation_records")
op.drop_column("generation_records", "pipeline_stage")
op.drop_column("generation_records", "video_upscale_snapshot_json")
op.drop_column("generation_records", "video_upscale_enabled_snapshot")
op.drop_column("generation_records", "provider_generation_resolution")
op.drop_column("chat_generation_tasks", "video_upscale_snapshot_json")
op.drop_column("chat_generation_tasks", "video_upscale_enabled_snapshot")
op.drop_column("chat_generation_tasks", "provider_generation_resolution")
video-gen-api/app/api/admin/__init__.py
+2
View File
@@ -1,6 +1,7 @@
from fastapi import APIRouter
from app.api.admin.video_prompt_schema_config import router as video_prompt_schema_config_router
from app.api.admin.video_upscale import router as video_upscale_router
from app.api.admin.resource_capacity import router as resource_capacity_router
from app.api.admin.team import router as team_router
from app.api.admin.home_material import router as home_material_router
@@ -11,6 +12,7 @@ from app.api.admin.upload import router as admin_upload_router
router = APIRouter()
router.include_router(video_prompt_schema_config_router)
router.include_router(video_upscale_router)
router.include_router(resource_capacity_router)
router.include_router(team_router)
router.include_router(home_material_router)
video-gen-api/app/api/admin/video_upscale.py
+53
View File
@@ -0,0 +1,53 @@
from __future__ import annotations
import json
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_admin_user, get_db
from app.models.user import User
from app.schemas.video_upscale import VideoUpscaleConfigOut, VideoUpscaleConfigSaveRequest
from app.services.operation_log import log_operation
from app.services.video_upscale.config_service import get_video_upscale_config, save_video_upscale_config
router = APIRouter(prefix="/admin/video-upscale", tags=["admin-video-upscale"])
@router.get("/config", response_model=VideoUpscaleConfigOut, summary="获取视频超分配置")
async def get_config(
admin: User = Depends(get_admin_user),
db: AsyncSession = Depends(get_db),
):
_ = admin
return await get_video_upscale_config(db)
@router.put("/config", response_model=VideoUpscaleConfigOut, summary="保存视频超分配置")
async def save_config(
req: VideoUpscaleConfigSaveRequest,
admin: User = Depends(get_admin_user),
db: AsyncSession = Depends(get_db),
):
before = await get_video_upscale_config(db)
result = await save_video_upscale_config(db, req.data)
await log_operation(
db,
admin.id,
admin.username,
"保存视频超分配置",
"PUT",
"/admin/video-upscale/config",
detail=json.dumps(
{
"before_version": before["data"].get("version"),
"after_version": result["data"].get("version"),
"enabled": result["data"].get("enabled"),
"delete_source_after_success": result["data"].get("delete_source_after_success"),
"rule_count": len(result["data"].get("rules") or []),
},
ensure_ascii=False,
),
)
await db.commit()
return result
video-gen-api/app/api/v1/admin.py
+31 -2
View File
@@ -24,6 +24,7 @@ from app.models.credit_ratio import CreditRatio
from app.models.operation_log import OperationLog
from app.enums.user import FrontendUserKind, UserType
from app.enums.team import TEAM_UNASSIGNED_VALUE
from app.enums.generation_status import GenerationRecordPipelineStage
from app.schemas.admin import (
CreditAdjustRequest,
ModelConfigCreate,
@@ -1920,6 +1921,8 @@ async def admin_list_generation_records(
"aspect_ratio": record.aspect_ratio,
"resolution": record.resolution,
"status": record.status,
"pipeline_stage": record.pipeline_stage,
"video_upscale_enabled": bool(record.video_upscale_enabled_snapshot),
"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 '',
"references": refs,
@@ -2033,6 +2036,8 @@ async def admin_generate_video(
if record.status not in ("prompt_optimized", "failed"):
raise HTTPException(status_code=400, detail=f"当前状态不允许生成{type_str}")
if record.pipeline_stage == GenerationRecordPipelineStage.UPSCALE_FAILED.value:
raise HTTPException(status_code=409, detail="该任务为画质增强失败，请使用超分恢复命令处理")
attempt_no = await get_next_credit_attempt_no(
db,
@@ -2049,6 +2054,21 @@ async def admin_generate_video(
if resolution not in RESOLUTIONS:
raise HTTPException(status_code=400, detail="不支持的分辨率")
from app.services.video_gen import get_active_engine
from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot
engine = await get_active_engine(db)
try:
supported_provider_resolutions = json.loads(engine.supported_resolutions or "[]")
except (TypeError, json.JSONDecodeError):
supported_provider_resolutions = []
provider_resolution, upscale_enabled, upscale_snapshot_json = await build_video_upscale_snapshot(
db,
target_resolution=resolution,
aspect_ratio=aspect_ratio,
supported_provider_resolutions=supported_provider_resolutions,
)
duration = record.duration or 5
media_billing = await charge_generation_media_by_params(
db,
@@ -2065,6 +2085,10 @@ async def admin_generate_video(
record.aspect_ratio = aspect_ratio
record.resolution = resolution
record.provider_generation_resolution = provider_resolution
record.video_upscale_enabled_snapshot = upscale_enabled
record.video_upscale_snapshot_json = upscale_snapshot_json
record.pipeline_stage = GenerationRecordPipelineStage.CREATING_PROVIDER_TASK.value
record.credits_cost = round(float(record.credits_cost or 0) + media_billing.total_charged, 2)
record.status = "generating"
record.error_message = None
@@ -2075,8 +2099,7 @@ async def admin_generate_video(
await db.flush()
try:
from app.services.video_gen import get_active_engine, submit_video_task
engine = await get_active_engine(db)
from app.services.video_gen import submit_video_task
task_id = await submit_video_task(
db,
engine,
@@ -2084,9 +2107,11 @@ async def admin_generate_video(
include_media_references=False,
)
record.seedance_task_id = task_id
record.pipeline_stage = GenerationRecordPipelineStage.WAITING_REMOTE.value
await db.flush()
await task_queue.enqueue(record_id)
except Exception as e:
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(
db,
record=record,
@@ -2119,6 +2144,10 @@ async def admin_generate_video(
record.video_url = None
record.video_cover_url = None
record.seedance_task_id = None
record.provider_generation_resolution = None
record.video_upscale_enabled_snapshot = False
record.video_upscale_snapshot_json = None
record.pipeline_stage = None
await db.flush()
try:
video-gen-api/app/api/v1/generation.py
+104 -58
View File
@@ -41,6 +41,7 @@ from app.services.resource_capacity_service import assert_user_resource_capacity
from app.services.upload_resource import delete_unbound_upload_resource, upload_reference_file, cleanup_upload_resource_files_after_commit
from app.services.upload_resource.log_service import log_upload_resource_exception, safe_rollback_with_log
from app.enums.upload_resource import UploadResourceEventEnum, UploadResourceModuleEnum, UploadResourceTypeEnum
from app.enums.generation_status import GenerationRecordPipelineStage
from app.services.generation.billing_service import (
CHARGE_TEXT_PROMPT,
OWNER_GENERATION_RECORD,
@@ -101,6 +102,8 @@ def _record_to_out(record: GenerationRecord, project_name: str, refs_override: l
image_proportion=record.image_proportion,
image_px=record.image_px,
status=record.status,
pipeline_stage=record.pipeline_stage,
video_upscale_enabled=bool(record.video_upscale_enabled_snapshot),
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 '',
@@ -416,6 +419,8 @@ async def generate(
record, project_name = row
if record.status not in ("prompt_optimized", "failed"):
raise InvalidStatusError("当前状态不允许生成")
if record.pipeline_stage == GenerationRecordPipelineStage.UPSCALE_FAILED.value:
raise InvalidStatusError("该任务生成失败，请联系客服进行修复")
await assert_user_resource_capacity_available(db, current_user.id)
@@ -432,6 +437,21 @@ async def generate(
if req.resolution not in RESOLUTIONS:
raise HTTPException(status_code=400, detail="不支持的分辨率")
from app.services.video_gen import get_active_engine
from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot
engine = await get_active_engine(db)
try:
supported_provider_resolutions = json.loads(engine.supported_resolutions or "[]")
except (TypeError, json.JSONDecodeError):
supported_provider_resolutions = []
provider_resolution, upscale_enabled, upscale_snapshot_json = await build_video_upscale_snapshot(
db,
target_resolution=req.resolution,
aspect_ratio=req.aspect_ratio,
supported_provider_resolutions=supported_provider_resolutions,
)
duration = record.duration or 5
media_billing = await charge_generation_media_by_params(
db,
@@ -448,6 +468,10 @@ async def generate(
record.aspect_ratio = req.aspect_ratio
record.resolution = req.resolution
record.provider_generation_resolution = provider_resolution
record.video_upscale_enabled_snapshot = upscale_enabled
record.video_upscale_snapshot_json = upscale_snapshot_json
record.pipeline_stage = GenerationRecordPipelineStage.CREATING_PROVIDER_TASK.value
record.credits_cost = round(float(record.credits_cost or 0) + media_billing.total_charged, 2)
record.status = "generating"
record.error_message = None
@@ -458,11 +482,10 @@ async def generate(
await db.flush()
try:
from app.services.video_gen import get_active_engine, submit_video_task
from app.services.video_gen import 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,
@@ -470,9 +493,11 @@ async def generate(
include_media_references=False,
)
record.seedance_task_id = task_id
record.pipeline_stage = GenerationRecordPipelineStage.WAITING_REMOTE.value
await db.flush()
await task_queue.enqueue(record_id)
except Exception as e:
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(
db,
record=record,
@@ -502,6 +527,10 @@ async def generate(
record.video_url = None
record.video_cover_url = None
record.seedance_task_id = None
record.provider_generation_resolution = None
record.video_upscale_enabled_snapshot = False
record.video_upscale_snapshot_json = None
record.pipeline_stage = None
await db.flush()
try:
@@ -543,6 +572,8 @@ async def retry_generation(
record, project_name = row
if record.status != "failed":
raise InvalidStatusError("只有失败的记录可以重试")
if record.pipeline_stage == GenerationRecordPipelineStage.UPSCALE_FAILED.value:
raise InvalidStatusError("该任务生成失败，请联系客服进行修复")
await assert_user_resource_capacity_available(db, current_user.id)
@@ -551,6 +582,27 @@ async def retry_generation(
owner_type=OWNER_GENERATION_RECORD,
owner_id=record.id,
)
engine = None
if record.gen_type == GenerationType.video:
from app.services.video_gen import get_active_engine
from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot
engine = await get_active_engine(db)
try:
supported_provider_resolutions = json.loads(engine.supported_resolutions or "[]")
except (TypeError, json.JSONDecodeError):
supported_provider_resolutions = []
provider_resolution, upscale_enabled, upscale_snapshot_json = await build_video_upscale_snapshot(
db,
target_resolution=record.resolution or "",
aspect_ratio=record.aspect_ratio or "",
supported_provider_resolutions=supported_provider_resolutions,
)
record.provider_generation_resolution = provider_resolution
record.video_upscale_enabled_snapshot = upscale_enabled
record.video_upscale_snapshot_json = upscale_snapshot_json
record.pipeline_stage = GenerationRecordPipelineStage.CREATING_PROVIDER_TASK.value
media_billing = await charge_generation_media_for_record(
db,
record=record,
@@ -572,8 +624,8 @@ async def retry_generation(
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)
from app.services.video_gen import submit_video_task
assert engine is not None
task_id = await submit_video_task(
db,
engine,
@@ -581,10 +633,13 @@ async def retry_generation(
include_media_references=False,
)
record.seedance_task_id = task_id
record.pipeline_stage = GenerationRecordPipelineStage.WAITING_REMOTE.value
await db.flush()
await task_queue.enqueue(record_id)
except Exception as e:
from app.services.error_codes import extract_error_message
if record.gen_type == GenerationType.video:
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(
db,
record=record,
@@ -679,6 +734,8 @@ async def get_queue_status(
return {
"record_id": record.id,
"status": record.status,
"pipeline_stage": record.pipeline_stage,
"video_upscale_enabled": bool(record.video_upscale_enabled_snapshot),
"queue_position": queue_position,
"estimated_wait_seconds": estimated_wait_seconds,
}
@@ -705,65 +762,54 @@ async def seedance_callback(request: Request, db: AsyncSession = Depends(get_db)
record = result.scalar_one_or_none()
if not record:
return {"message": "record not found"}
if record.status == "completed":
return {"message": "already completed"}
if str(record.pipeline_stage or "").startswith("upscale_"):
return {"message": "upscale already started"}
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(CST)
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,
remote_url = str(data.get("content", {}).get("video_url", "") or "").strip()
if not remote_url:
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(
db, record=record, error_message="供应商回调成功但未返回视频地址"
)
# 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)
else:
usage = data.get("usage", {}) if isinstance(data.get("usage"), dict) else {}
from app.services.video_queue import handle_generation_record_video_succeeded
try:
entered_upscale = await handle_generation_record_video_succeeded(
db,
record,
remote_url=remote_url,
provider_response=data,
video_tokens=usage.get("total_tokens", 0),
)
except Exception as exc:
await db.rollback()
result = await db.execute(
select(GenerationRecord).where(GenerationRecord.id == record.id).with_for_update().limit(1)
)
failed_record = result.scalar_one_or_none()
if failed_record:
failed_record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(
db, record=failed_record, error_message=f"视频结果下载失败: {exc}"
)
entered_upscale = False
from app.services.video_gen import _log_video_response
_log_video_response(record.id, data)
if not entered_upscale and record.status == "completed":
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", "视频生成失败")
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(
db,
record=record,
video-gen-api/app/api/v1/generation_ai.py
+14
View File
@@ -761,6 +761,20 @@ async def retry_task(
raise HTTPException(status_code=400, detail="只有失败任务可以重试")
retry_targets = [task]
upscale_failed_ids = [
str(target.id)
for target in retry_targets
if target.pipeline_stage == ChatGenerationPipelineStage.UPSCALE_FAILED.value
]
if upscale_failed_ids:
raise HTTPException(
status_code=409,
detail={
"message": "画质增强失败任务不能通过普通生成重试，请由管理员使用视频超分恢复命令处理",
"task_ids": upscale_failed_ids,
},
)
await assert_user_resource_capacity_available(db, current_user.id)
enqueue_ids: list[str] = []
download_retry_ids: list[str] = []
video-gen-api/app/api/v1/projects.py
+5
View File
@@ -10,6 +10,7 @@ from app.models.project import Project
from app.models.generation_record import GenerationRecord
from app.schemas.project import ProjectCreate, ProjectOut
from app.services.resource_accounting_service import soft_delete_generation_record_resources
from app.services.video_upscale.guard_service import assert_no_recoverable_failed_upscale_tasks
from app.utils.id_gen import generate_id
router = APIRouter(prefix="/projects", tags=["projects"])
@@ -88,6 +89,10 @@ async def delete_project(
)
records = list(records_result.scalars().all())
record_ids = [record.id for record in records]
await assert_no_recoverable_failed_upscale_tasks(
db,
generation_record_ids=record_ids,
)
now = datetime.now(timezone.utc)
project.deleted_at = now
video-gen-api/app/cmd/retry_video_upscale.py
+188
View File
@@ -0,0 +1,188 @@
from __future__ import annotations
import argparse
import asyncio
import json
from datetime import datetime, timezone
from sqlalchemy import or_, select
from app.config import settings
from app.enums.video_upscale import LOCAL_PROCESSOR_KEYS, VideoUpscaleTaskStatus
from app.models.base import async_session
from app.models.chat_generation_task import ChatGenerationTask
from app.models.generation_record import GenerationRecord
from app.models.module_generation_step import ModuleGenerationStep
from app.models.shot_replicate_segment import ShotReplicateSegment
from app.models.video_upscale_task import VideoUpscaleTask
from app.services.video_upscale.media_service import is_valid_file
from app.services.video_upscale.task_service import reset_failed_upscale_task_for_manual_retry
def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="人工恢复视频超分任务")
parser.add_argument("--task-id", action="append", default=[], help="ChatGenerationTask.id，可重复传入")
parser.add_argument("--task-ids", default="", help="逗号分隔的 ChatGenerationTask.id")
parser.add_argument("--generation-record-id", action="append", default=[], help="GenerationRecord.id，可重复传入")
parser.add_argument("--generation-record-ids", default="", help="逗号分隔的 GenerationRecord.id")
parser.add_argument("--project-id", action="append", default=[], help="Project.id，可重复传入")
parser.add_argument("--generation-mode", default="", help="按 ChatGenerationTask.generation_mode 筛选")
parser.add_argument("--module-owner-id", default="", help="ModuleGenerationProject.id")
parser.add_argument("--shot-task-set-id", default="", help="ShotReplicateTaskSet.id")
parser.add_argument("--shot-segment-id", action="append", default=[], help="ShotReplicateSegment.id，可重复传入")
parser.add_argument("--failed-only", action=argparse.BooleanOptionalAction, default=True)
parser.add_argument("--limit", type=int, default=100)
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--enqueue", action=argparse.BooleanOptionalAction, default=True)
parser.add_argument("--force-resubmit", action="store_true", help="远程任务清空 provider task/result 后从 source.mp4 重新提交")
return parser
async def _collect_chat_task_ids(db, args: argparse.Namespace) -> list[str]:
ids = [str(item).strip() for item in args.task_id if str(item).strip()]
ids.extend(item.strip() for item in str(args.task_ids or "").split(",") if item.strip())
project_ids: list[str] = [str(args.module_owner_id).strip()] if args.module_owner_id else []
segment_ids = [str(item).strip() for item in args.shot_segment_id if str(item).strip()]
if args.shot_task_set_id or segment_ids:
query = select(ShotReplicateSegment.module_project_id).where(
ShotReplicateSegment.deleted_at.is_(None),
ShotReplicateSegment.module_project_id.is_not(None),
)
if args.shot_task_set_id:
query = query.where(ShotReplicateSegment.task_set_id == str(args.shot_task_set_id).strip())
if segment_ids:
query = query.where(ShotReplicateSegment.id.in_(segment_ids))
result = await db.execute(query)
project_ids.extend(str(value) for value in result.scalars().all() if value)
project_ids = list(dict.fromkeys(item for item in project_ids if item))
if project_ids:
result = await db.execute(
select(ModuleGenerationStep.chat_task_id).where(
ModuleGenerationStep.project_id.in_(project_ids),
ModuleGenerationStep.chat_task_id.isnot(None),
ModuleGenerationStep.deleted_at.is_(None),
)
)
ids.extend(str(value) for value in result.scalars().all() if value)
return list(dict.fromkeys(ids))
async def _collect_generation_record_ids(db, args: argparse.Namespace) -> list[str]:
ids = [str(item).strip() for item in args.generation_record_id if str(item).strip()]
ids.extend(item.strip() for item in str(args.generation_record_ids or "").split(",") if item.strip())
project_ids = [str(item).strip() for item in args.project_id if str(item).strip()]
if project_ids:
result = await db.execute(
select(GenerationRecord.id).where(
GenerationRecord.project_id.in_(project_ids),
GenerationRecord.deleted_at.is_(None),
)
)
ids.extend(str(value) for value in result.scalars().all() if value)
return list(dict.fromkeys(ids))
def _owner_preview(upscale: VideoUpscaleTask, chat: ChatGenerationTask | None, record: GenerationRecord | None) -> dict:
owner = chat or record
return {
"upscale_task_id": upscale.id,
"owner_type": "chat_generation_task" if chat else "generation_record",
"owner_id": owner.id if owner else None,
"chat_task_id": chat.id if chat else None,
"generation_record_id": record.id if record else None,
"project_id": record.project_id if record else None,
"generation_mode": chat.generation_mode if chat else None,
"status": upscale.status,
"stage": upscale.stage,
"processor_key": upscale.processor_key,
"source_local_path": upscale.source_local_path,
"provider_task_id": upscale.provider_task_id,
"provider_output_url_expires_at": upscale.provider_output_url_expires_at,
}
async def _run(args: argparse.Namespace) -> dict:
async with async_session() as db:
chat_ids = await _collect_chat_task_ids(db, args)
record_ids = await _collect_generation_record_ids(db, args)
query = (
select(VideoUpscaleTask, ChatGenerationTask, GenerationRecord)
.outerjoin(ChatGenerationTask, ChatGenerationTask.id == VideoUpscaleTask.chat_generation_task_id)
.outerjoin(GenerationRecord, GenerationRecord.id == VideoUpscaleTask.generation_record_id)
.where(
or_(
(ChatGenerationTask.id.isnot(None) & ChatGenerationTask.deleted_at.is_(None)),
(GenerationRecord.id.isnot(None) & GenerationRecord.deleted_at.is_(None)),
)
)
.order_by(VideoUpscaleTask.updated_at.asc())
.limit(max(1, min(int(args.limit or 100), 1000)))
)
owner_filters = []
if chat_ids:
owner_filters.append(VideoUpscaleTask.chat_generation_task_id.in_(chat_ids))
if record_ids:
owner_filters.append(VideoUpscaleTask.generation_record_id.in_(record_ids))
if owner_filters:
query = query.where(or_(*owner_filters))
if args.generation_mode:
query = query.where(ChatGenerationTask.generation_mode == args.generation_mode)
if args.failed_only:
query = query.where(VideoUpscaleTask.status == VideoUpscaleTaskStatus.FAILED.value)
result = await db.execute(query)
rows = result.all()
preview = [_owner_preview(upscale, chat, record) for upscale, chat, record in rows]
if args.dry_run or not args.enqueue:
return {"dry_run": True, "matched": len(preview), "items": preview}
from app.tasks.video_upscale_tasks import download_remote_result, execute_local, finalize, poll_remote, submit_remote
enqueued = []
for upscale, chat, record in rows:
reset = await reset_failed_upscale_task_for_manual_retry(
db,
upscale_task_id=upscale.id,
force_resubmit=bool(args.force_resubmit),
)
if is_valid_file(reset.final_local_path) and not args.force_resubmit:
action = "finalize"
finalize.apply_async(args=[reset.id], queue=settings.VIDEO_UPSCALE_LOCAL_QUEUE)
elif reset.processor_key in LOCAL_PROCESSOR_KEYS:
action = "local"
execute_local.apply_async(args=[reset.id], queue=settings.VIDEO_UPSCALE_LOCAL_QUEUE)
else:
expires_at = reset.provider_output_url_expires_at
if expires_at and expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=timezone.utc)
remaining = (expires_at - datetime.now(timezone.utc)).total_seconds() if expires_at else None
if reset.provider_output_url and remaining is not None and remaining >= 2 * 3600 and not args.force_resubmit:
action = "download"
download_remote_result.apply_async(args=[reset.id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE)
elif reset.provider_task_id and not args.force_resubmit:
action = "poll"
poll_remote.apply_async(args=[reset.id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE)
else:
action = "submit"
submit_remote.apply_async(args=[reset.id], queue=settings.VIDEO_UPSCALE_REMOTE_QUEUE)
owner = chat or record
enqueued.append(
{
"upscale_task_id": reset.id,
"owner_type": "chat_generation_task" if chat else "generation_record",
"owner_id": owner.id if owner else None,
"action": action,
}
)
return {"dry_run": False, "matched": len(preview), "enqueued": enqueued}
def main() -> None:
args = _parser().parse_args()
result = asyncio.run(_run(args))
print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
if __name__ == "__main__":
main()
video-gen-api/app/config.py
+22
View File
@@ -95,6 +95,28 @@ class Settings(BaseSettings):
# - FFMPEG_BIN 为空时自动从系统 PATH 查找 ffmpeg / ffmpeg.exe。
# - VIDEO_COVER_TIMEOUT_SECONDS 必须较短，避免 ffmpeg 异常卡住下载 worker。
FFMPEG_BIN: str = ""
# 视频超分配置。
# 本地处理器始终复用 FFMPEG_BIN，不允许由管理后台覆盖可执行文件路径。
VOLC_API_KEY: str = ""
VOLC_MEDIAKIT_API_BASE: str = "https://mediakit.cn-beijing.volces.com"
VIDEO_UPSCALE_LOCAL_QUEUE: str = "gen_video_upscale_local"
VIDEO_UPSCALE_REMOTE_QUEUE: str = "gen_video_upscale_remote"
VIDEO_UPSCALE_LOCAL_TIMEOUT_SECONDS: int = 3600
VIDEO_UPSCALE_MAX_ATTEMPTS: int = 3
VIDEO_UPSCALE_REMOTE_POLL_INTERVAL_SECONDS: int = 30
VIDEO_UPSCALE_REMOTE_POLL_TIMEOUT_SECONDS: int = 7200
VIDEO_UPSCALE_REMOTE_REQUEST_TIMEOUT_SECONDS: int = 30
VIDEO_UPSCALE_REMOTE_RESULT_DOWNLOAD_TIMEOUT_SECONDS: int = 600
VIDEO_UPSCALE_REMOTE_URL_PROBE_THRESHOLD_SECONDS: int = 600
VIDEO_UPSCALE_REMOTE_URL_PROBE_CONNECT_TIMEOUT_SECONDS: int = 3
VIDEO_UPSCALE_REMOTE_URL_PROBE_READ_TIMEOUT_SECONDS: int = 5
VIDEO_UPSCALE_LOCAL_SIGNED_URL_EXPIRE_SECONDS: int = 7200
VIDEO_UPSCALE_TASK_LEASE_SECONDS: int = 30 * 60
VIDEO_UPSCALE_RETRY_BACKOFF_SECONDS: int = 60
VIDEO_UPSCALE_RECOVERY_BATCH_SIZE: int = 50
VIDEO_UPSCALE_RECOVERY_LOCK_KEY: str = "vg:celery:video_upscale_recovery_lock"
VIDEO_COVER_SEEK_TIME: str = "00:00:01"
VIDEO_COVER_FALLBACK_SEEK_TIME: str = "00:00:00"
VIDEO_COVER_WIDTH: int = 720
video-gen-api/app/enums/celery_queue.py
+8
View File
@@ -5,6 +5,8 @@ class CeleryQueue(str, Enum):
GEN_CHATAPI_CREATE = "gen_chatapi_create"
GEN_PROVIDER_POLL = "gen_provider_poll"
GEN_RESULT_DOWNLOAD = "gen_result_download"
GEN_VIDEO_UPSCALE_LOCAL = "gen_video_upscale_local"
GEN_VIDEO_UPSCALE_REMOTE = "gen_video_upscale_remote"
GEN_RECOVERY = "gen_recovery"
GEN_PRIVATE_PORTRAIT = "gen_private_portrait"
DEFAULT = "default"
@@ -16,6 +18,12 @@ class CeleryTaskName(str, Enum):
DOWNLOAD_GENERATION_RESULT = "generation.download_generation_result_task"
RECOVER_DOWNLOAD = "generation.recover_download_tasks_once"
RECOVER_GENERATION = "generation.recover_generation_tasks_once"
VIDEO_UPSCALE_EXECUTE_LOCAL = "video_upscale.execute_local"
VIDEO_UPSCALE_SUBMIT_REMOTE = "video_upscale.submit_remote"
VIDEO_UPSCALE_POLL_REMOTE = "video_upscale.poll_remote"
VIDEO_UPSCALE_DOWNLOAD_REMOTE_RESULT = "video_upscale.download_remote_result"
VIDEO_UPSCALE_FINALIZE = "video_upscale.finalize"
VIDEO_UPSCALE_RECOVER = "video_upscale.recover_once"
DISPATCH_DUE_POLL = "generation.dispatch_due_poll_tasks"
STARTUP_RECOVERY = "recovery.startup_recovery_once"
MODULE_ASYNC_RECOVERY = "module_async.recover_module_async_tasks_once"
video-gen-api/app/enums/generation_status.py
+19
View File
@@ -9,6 +9,25 @@ class GenerationStatus(str, Enum):
failed = "failed"
class GenerationRecordPipelineStage(str, Enum):
"""GenerationRecord 视频生成与超分流水线阶段。"""
CREATING_PROVIDER_TASK = "creating_provider_task"
WAITING_REMOTE = "waiting_remote"
POLLING = "polling"
RESULT_READY = "result_ready"
DOWNLOADING = "downloading"
UPSCALE_QUEUED = "upscale_queued"
UPSCALE_PROCESSING = "upscale_processing"
UPSCALE_POLLING = "upscale_polling"
UPSCALE_DOWNLOADING = "upscale_downloading"
UPSCALE_FINALIZING = "upscale_finalizing"
UPSCALE_RETRY_WAITING = "upscale_retry_waiting"
UPSCALE_FAILED = "upscale_failed"
DONE = "done"
FAILED = "failed"
class GenerationType(str, Enum):
"""生成类型。"""
video = "video"
video-gen-api/app/enums/generation_task.py
+21
View File
@@ -40,6 +40,13 @@ class ChatGenerationPipelineStage(str, Enum):
DOWNLOAD_QUEUED = "download_queued"
DOWNLOADING = "downloading"
RETRY_WAITING = "retry_waiting"
UPSCALE_QUEUED = "upscale_queued"
UPSCALE_PROCESSING = "upscale_processing"
UPSCALE_POLLING = "upscale_polling"
UPSCALE_DOWNLOADING = "upscale_downloading"
UPSCALE_FINALIZING = "upscale_finalizing"
UPSCALE_RETRY_WAITING = "upscale_retry_waiting"
UPSCALE_FAILED = "upscale_failed"
DONE = "done"
FAILED = "failed"
TIMEOUT = "timeout"
@@ -108,6 +115,19 @@ class ChatGenerationTaskEventType(str, Enum):
DOWNLOAD_FAILED = "DOWNLOAD_FAILED"
DOWNLOAD_FAILED_NON_RETRYABLE = "DOWNLOAD_FAILED_NON_RETRYABLE"
UPSCALE_SNAPSHOT_MATCHED = "UPSCALE_SNAPSHOT_MATCHED"
UPSCALE_SNAPSHOT_BYPASSED = "UPSCALE_SNAPSHOT_BYPASSED"
UPSCALE_SOURCE_READY = "UPSCALE_SOURCE_READY"
UPSCALE_ENQUEUE = "UPSCALE_ENQUEUE"
UPSCALE_START = "UPSCALE_START"
UPSCALE_REMOTE_SUBMIT = "UPSCALE_REMOTE_SUBMIT"
UPSCALE_REMOTE_POLL = "UPSCALE_REMOTE_POLL"
UPSCALE_REMOTE_RESULT_READY = "UPSCALE_REMOTE_RESULT_READY"
UPSCALE_RETRY_WAITING = "UPSCALE_RETRY_WAITING"
UPSCALE_SUCCESS = "UPSCALE_SUCCESS"
UPSCALE_FAILED = "UPSCALE_FAILED"
UPSCALE_RECOVERY_ENQUEUE = "UPSCALE_RECOVERY_ENQUEUE"
DOWNLOAD_SKIP_TASK_MISSING = "DOWNLOAD_SKIP_TASK_MISSING"
DOWNLOAD_SKIP_INVALID_MODE = "DOWNLOAD_SKIP_INVALID_MODE"
DOWNLOAD_SKIP_NOT_GENERATING = "DOWNLOAD_SKIP_NOT_GENERATING"
@@ -149,6 +169,7 @@ FINAL_CHAT_GENERATION_STAGES = {
ChatGenerationPipelineStage.FAILED.value,
ChatGenerationPipelineStage.TIMEOUT.value,
ChatGenerationPipelineStage.DOWNLOAD_FAILED.value,
ChatGenerationPipelineStage.UPSCALE_FAILED.value,
}
DOWNLOAD_RECOVERABLE_STAGES = {
video-gen-api/app/enums/video_upscale.py
+97
View File
@@ -0,0 +1,97 @@
from __future__ import annotations
from enum import Enum
VIDEO_UPSCALE_CONFIG_KEY = "video_upscale_config"
VIDEO_UPSCALE_CONFIG_DESCRIPTION = "视频生成超分全局配置"
VIDEO_UPSCALE_CONFIG_VERSION = 1
VIDEO_UPSCALE_SOURCE_RETAINED_MARKER = "retained_by_snapshot_config"
class VideoUpscaleProcessorKey(str, Enum):
LOCAL_FFMPEG_CROP_V1 = "local_ffmpeg_crop_v1"
VOLC_LARGE_MODEL_V1 = "volc_large_model_v1"
VOLC_STANDARD_V1 = "volc_standard_v1"
VOLC_PROFESSIONAL_V1 = "volc_professional_v1"
class VideoUpscaleTaskStatus(str, Enum):
PENDING = "pending"
PROCESSING = "processing"
RETRY_WAITING = "retry_waiting"
COMPLETED = "completed"
FAILED = "failed"
class VideoUpscaleStage(str, Enum):
QUEUED = "upscale_queued"
SOURCE_READY = "upscale_source_ready"
LOCAL_PROCESSING = "upscale_local_processing"
REMOTE_SUBMITTING = "upscale_remote_submitting"
REMOTE_POLLING = "upscale_remote_polling"
RESULT_READY = "upscale_result_ready"
RESULT_DOWNLOADING = "upscale_result_downloading"
VALIDATING = "upscale_validating"
GENERATING_COVER = "upscale_generating_cover"
FINALIZING = "upscale_finalizing"
RETRY_WAITING = "upscale_retry_waiting"
COMPLETED = "upscale_completed"
FAILED = "upscale_failed"
class VideoUpscaleInputSourceType(str, Enum):
PROVIDER_REMOTE = "provider_remote"
LOCAL_SIGNED = "local_signed"
class VideoUpscaleProbeStatus(str, Enum):
NOT_CHECKED = "not_checked"
SUCCESS = "success"
FAILED = "failed"
EXPIRED = "expired"
UNPARSABLE = "unparsable"
VIDEO_UPSCALE_RESOLUTIONS = ("480p", "720p", "1080p", "2K", "4K")
VIDEO_UPSCALE_RESOLUTION_SHORT_EDGE = {
"480p": 480,
"720p": 720,
"1080p": 1080,
"2K": 1440,
"4K": 2160,
}
VIDEO_UPSCALE_RESOLUTION_RANK = {
resolution: index for index, resolution in enumerate(VIDEO_UPSCALE_RESOLUTIONS, start=1)
}
def normalize_video_upscale_resolution(value: str | None) -> str:
text = str(value or "").strip()
normalized = text.lower()
aliases = {
"480p": "480p",
"720p": "720p",
"1080p": "1080p",
"2k": "2K",
"4k": "4K",
}
return aliases.get(normalized, text)
def video_upscale_short_edge_pixels(value: str | None) -> int:
normalized = normalize_video_upscale_resolution(value)
try:
return VIDEO_UPSCALE_RESOLUTION_SHORT_EDGE[normalized]
except KeyError as exc:
supported = "、".join(VIDEO_UPSCALE_RESOLUTIONS)
raise ValueError(f"不支持的视频超分分辨率: {value}，仅支持 {supported}") from exc
LOCAL_PROCESSOR_KEYS = {VideoUpscaleProcessorKey.LOCAL_FFMPEG_CROP_V1.value}
REMOTE_PROCESSOR_KEYS = {
VideoUpscaleProcessorKey.VOLC_LARGE_MODEL_V1.value,
VideoUpscaleProcessorKey.VOLC_STANDARD_V1.value,
VideoUpscaleProcessorKey.VOLC_PROFESSIONAL_V1.value,
}
ALL_PROCESSOR_KEYS = LOCAL_PROCESSOR_KEYS | REMOTE_PROCESSOR_KEYS
video-gen-api/app/models/__init__.py
+2 -1
View File
@@ -21,6 +21,7 @@ from app.models.operation_log import OperationLog
from app.models.chat_generation_task import ChatGenerationTask
from app.models.chat_generation_task_event import ChatGenerationTaskEvent
from app.models.chat_provider_call_log import ChatProviderCallLog
from app.models.video_upscale_task import VideoUpscaleTask
from app.models.generated_resource import GeneratedResource
from app.models.upload_resource import UploadResource
from app.models.user_resource_month_stat import UserResourceMonthStat
@@ -44,7 +45,7 @@ __all__ = [
"ModelConfig", "SystemConfig", "Notification", "PaymentOrder",
"TokenUsage", "IndustryConfig", "VideoEngine", "CreditRatio",
"MenuConfig", "RechargePackage", "OperationLog", "ContactRequest",
"ChatGenerationTask", "ChatGenerationTaskEvent", "ChatProviderCallLog",
"ChatGenerationTask", "ChatGenerationTaskEvent", "ChatProviderCallLog", "VideoUpscaleTask",
"GeneratedResource", "UploadResource", "UserResourceMonthStat", "UserResourceTotalStat",
"UserResourceCapacityConfig",
"ModuleGenerationProject", "ModuleGenerationStep",
video-gen-api/app/models/chat_generation_task.py
+4 -1
View File
@@ -1,6 +1,6 @@
from datetime import datetime
from sqlalchemy import CheckConstraint, DateTime, Float, ForeignKey, Index, Integer, String, Text, text
from sqlalchemy import Boolean, CheckConstraint, DateTime, Float, ForeignKey, Index, Integer, String, Text, text
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base, TimestampMixin, SoftDeleteMixin
@@ -76,6 +76,9 @@ class ChatGenerationTask(Base, TimestampMixin, SoftDeleteMixin):
duration: Mapped[int | None] = mapped_column(Integer, nullable=True)
aspect_ratio: Mapped[str | None] = mapped_column(String(8), nullable=True)
resolution: Mapped[str | None] = mapped_column(String(8), nullable=True)
provider_generation_resolution: Mapped[str | None] = mapped_column(String(16), nullable=True)
video_upscale_enabled_snapshot: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="false")
video_upscale_snapshot_json: Mapped[str | None] = mapped_column(Text, nullable=True)
image_size: Mapped[str | None] = mapped_column(String(16), nullable=True)
image_proportion: Mapped[str | None] = mapped_column(String(8), nullable=True)
image_px: Mapped[str | None] = mapped_column(String(16), nullable=True)
video-gen-api/app/models/generation_record.py
+7 -1
View File
@@ -1,6 +1,6 @@
from datetime import datetime
from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, Float, Index
from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, Float, Index
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base, TimestampMixin, SoftDeleteMixin
@@ -22,11 +22,17 @@ class GenerationRecord(Base, TimestampMixin, SoftDeleteMixin):
duration: Mapped[int | None] = mapped_column(Integer, nullable=True)
aspect_ratio: Mapped[str | None] = mapped_column(String(8), nullable=True)
resolution: Mapped[str | None] = mapped_column(String(8), nullable=True)
provider_generation_resolution: Mapped[str | None] = mapped_column(String(16), nullable=True)
video_upscale_enabled_snapshot: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=False, server_default="false"
)
video_upscale_snapshot_json: Mapped[str | None] = mapped_column(Text, nullable=True)
image_size: Mapped[str | None] = mapped_column(String(8), nullable=True)
image_proportion: Mapped[str | None] = mapped_column(String(8), nullable=True)
image_px: Mapped[str | None] = mapped_column(String(10), nullable=True)
status: Mapped[str] = mapped_column(String(32), default="prompt_optimized")
pipeline_stage: Mapped[str | None] = mapped_column(String(48), nullable=True, index=True)
video_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
video_cover_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
image_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
video-gen-api/app/models/video_upscale_task.py
+85
View File
@@ -0,0 +1,85 @@
from __future__ import annotations
from datetime import datetime
from sqlalchemy import BigInteger, CheckConstraint, DateTime, Float, ForeignKey, Index, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.models.base import Base, TimestampMixin
class VideoUpscaleTask(Base, TimestampMixin):
__tablename__ = "video_upscale_tasks"
__table_args__ = (
CheckConstraint(
"(chat_generation_task_id IS NOT NULL AND generation_record_id IS NULL) OR "
"(chat_generation_task_id IS NULL AND generation_record_id IS NOT NULL)",
name="ck_video_upscale_tasks_exactly_one_owner",
),
Index("uq_video_upscale_tasks_chat_task", "chat_generation_task_id", unique=True),
Index("uq_video_upscale_tasks_generation_record", "generation_record_id", unique=True),
Index("idx_video_upscale_tasks_provider_task_id", "provider_task_id"),
Index("idx_video_upscale_tasks_status_next_retry", "status", "next_retry_at"),
Index("idx_video_upscale_tasks_status_lease", "status", "lease_until"),
)
id: Mapped[str] = mapped_column(String(32), primary_key=True)
chat_generation_task_id: Mapped[str | None] = mapped_column(
String(32),
ForeignKey("chat_generation_tasks.id", ondelete="CASCADE"),
nullable=True,
)
generation_record_id: Mapped[str | None] = mapped_column(
String(32),
ForeignKey("generation_records.id", ondelete="CASCADE"),
nullable=True,
)
status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending", server_default="pending")
stage: Mapped[str] = mapped_column(String(48), nullable=False, default="upscale_queued", server_default="upscale_queued")
processor_key: Mapped[str] = mapped_column(String(64), nullable=False)
attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
failure_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
manual_retry_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
next_retry_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
source_local_path: Mapped[str | None] = mapped_column(Text, nullable=True)
source_file_size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0, server_default="0")
source_width: Mapped[int | None] = mapped_column(Integer, nullable=True)
source_height: Mapped[int | None] = mapped_column(Integer, nullable=True)
source_duration_seconds: Mapped[float | None] = mapped_column(Float, nullable=True)
source_deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
source_delete_error: Mapped[str | None] = mapped_column(Text, nullable=True)
source_remote_url: Mapped[str | None] = mapped_column(Text, nullable=True)
source_remote_url_signed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
source_remote_url_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
source_remote_url_last_probe_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
source_remote_url_probe_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
input_source_type: Mapped[str | None] = mapped_column(String(32), nullable=True)
input_source_fallback_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")
target_width: Mapped[int] = mapped_column(Integer, nullable=False)
target_height: Mapped[int] = mapped_column(Integer, nullable=False)
effective_target_width: Mapped[int | None] = mapped_column(Integer, nullable=True)
effective_target_height: Mapped[int | None] = mapped_column(Integer, nullable=True)
provider_task_id: Mapped[str | None] = mapped_column(String(160), nullable=True)
provider_request_json: Mapped[str | None] = mapped_column(Text, nullable=True)
provider_response_json: Mapped[str | None] = mapped_column(Text, nullable=True)
provider_output_url: Mapped[str | None] = mapped_column(Text, nullable=True)
provider_output_url_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
provider_submitted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
final_local_path: Mapped[str | None] = mapped_column(Text, nullable=True)
final_resource_url: Mapped[str | None] = mapped_column(Text, nullable=True)
final_file_size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0, server_default="0")
celery_task_id: Mapped[str | None] = mapped_column(String(160), nullable=True)
lease_token: Mapped[str | None] = mapped_column(String(64), nullable=True)
lease_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
failed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
video-gen-api/app/schemas/generation.py
+2
View File
@@ -50,6 +50,8 @@ class GenerationRecordOut(BaseModel):
image_proportion: str | None = None
image_px: str | None = None
status: str
pipeline_stage: str | None = None
video_upscale_enabled: bool = False
video_url: str | None = None
video_cover_url: str | None = None
image_url: str | None = None
video-gen-api/app/schemas/video_upscale.py
+59
View File
@@ -0,0 +1,59 @@
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from app.enums.video_upscale import normalize_video_upscale_resolution
from app.schemas.common import NaiveDatetimeOptional
class VideoUpscaleResolutionRule(BaseModel):
model_config = ConfigDict(extra="forbid")
target_resolution: str = Field(..., min_length=1, max_length=16)
provider_generation_resolution: str = Field(..., min_length=1, max_length=16)
processor_key: str = Field(..., min_length=1, max_length=64)
enabled: bool = True
@field_validator("target_resolution", "provider_generation_resolution")
@classmethod
def clean_resolution(cls, value: str) -> str:
return normalize_video_upscale_resolution(value)
@field_validator("processor_key")
@classmethod
def clean_processor_key(cls, value: str) -> str:
return str(value).strip()
class VideoUpscaleConfigData(BaseModel):
model_config = ConfigDict(extra="forbid")
enabled: bool = False
version: int = Field(1, ge=1)
delete_source_after_success: bool = True
rules: list[VideoUpscaleResolutionRule] = Field(default_factory=list)
@model_validator(mode="after")
def validate_unique_rules(self) -> "VideoUpscaleConfigData":
seen: set[str] = set()
for rule in self.rules:
if not rule.enabled:
continue
if rule.target_resolution in seen:
raise ValueError(f"客户目标分辨率存在重复启用规则: {rule.target_resolution}")
seen.add(rule.target_resolution)
return self
class VideoUpscaleConfigSaveRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
data: VideoUpscaleConfigData
class VideoUpscaleConfigOut(BaseModel):
id: str | None = None
key: str
description: str | None = None
data: VideoUpscaleConfigData
created_at: NaiveDatetimeOptional = None
updated_at: NaiveDatetimeOptional = None
video-gen-api/app/services/generation/ai/task_create_service.py
+20
View File
@@ -40,6 +40,7 @@ from app.services.generation.billing_service import OWNER_CHAT_GENERATION_TASK,
from app.services.operation_log_service import log_operation_event
from app.services.private_portrait.reference_resolver import resolve_private_portrait_references
from app.services.resource_capacity_service import assert_user_resource_capacity_available
from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot
from app.utils.id_gen import generate_id
@@ -331,7 +332,15 @@ async def create_generation_task_group(
raise HTTPException(status_code=400, detail=f"视频时长不能超过 {engine.max_duration} 秒")
input_video_duration = _validate_video_references(refs, max_audio_count=engine.max_audio_count)
provider_generation_resolution, upscale_enabled_snapshot, upscale_snapshot_json = await build_video_upscale_snapshot(
db,
target_resolution=resolution,
aspect_ratio=ratio,
supported_provider_resolutions=resolutions,
)
video_snapshot = build_video_snapshot(engine, ratio, resolution, duration)
video_snapshot["provider_generation_resolution"] = provider_generation_resolution
video_snapshot["video_upscale_enabled_snapshot"] = upscale_enabled_snapshot
video_snapshot["generation_count"] = generation_count
snapshot_json = _json(video_snapshot) or "{}"
deadline_at = now + timedelta(hours=settings.CHATAPI_ASYNC_VIDEO_FINAL_DEADLINE_HOURS)
@@ -370,6 +379,9 @@ async def create_generation_task_group(
duration=duration,
aspect_ratio=ratio,
resolution=resolution,
provider_generation_resolution=provider_generation_resolution,
video_upscale_enabled_snapshot=upscale_enabled_snapshot,
video_upscale_snapshot_json=upscale_snapshot_json,
image_size=req.image_size or IMAGE_DEFAULT_SIZE,
image_proportion=req.image_proportion or IMAGE_DEFAULT_PROPORTION,
image_px=normalize_px(req.image_px) or IMAGE_DEFAULT_PX,
@@ -394,6 +406,9 @@ async def create_generation_task_group(
duration=duration,
aspect_ratio=ratio,
resolution=resolution,
provider_generation_resolution=provider_generation_resolution,
video_upscale_enabled_snapshot=upscale_enabled_snapshot,
video_upscale_snapshot_json=upscale_snapshot_json,
image_size=req.image_size or IMAGE_DEFAULT_SIZE,
image_proportion=req.image_proportion or IMAGE_DEFAULT_PROPORTION,
image_px=normalize_px(req.image_px) or IMAGE_DEFAULT_PX,
@@ -438,6 +453,9 @@ async def create_generation_task_group(
duration=duration,
aspect_ratio=ratio,
resolution=resolution,
provider_generation_resolution=provider_generation_resolution,
video_upscale_enabled_snapshot=upscale_enabled_snapshot,
video_upscale_snapshot_json=upscale_snapshot_json,
image_size=req.image_size or IMAGE_DEFAULT_SIZE,
image_proportion=req.image_proportion or IMAGE_DEFAULT_PROPORTION,
image_px=normalize_px(req.image_px) or IMAGE_DEFAULT_PX,
@@ -476,6 +494,8 @@ async def create_generation_task_group(
"generation_count": generation_count,
"child_task_ids": child_ids,
"enqueue_task_ids": enqueue_ids,
"video_upscale_enabled_snapshot": bool(locals().get("upscale_enabled_snapshot", False)),
"provider_generation_resolution": locals().get("provider_generation_resolution"),
},
)
return GenerationTaskCreateResult(
video-gen-api/app/services/generation/ai/task_group_service.py
+20 -5
View File
@@ -15,6 +15,7 @@ from app.enums.generation_task import (
)
from app.models.chat_generation_task import ChatGenerationTask
from app.services.operation_log_service import log_operation_event
from app.services.video_upscale.guard_service import assert_no_recoverable_failed_upscale_tasks
from app.services.resource_accounting_service import (
SOURCE_MODEL_CHAT_TASK,
soft_delete_resources_by_source,
@@ -31,6 +32,12 @@ ACTIVE_STAGES = {
ChatGenerationPipelineStage.DOWNLOAD_QUEUED.value,
ChatGenerationPipelineStage.DOWNLOADING.value,
ChatGenerationPipelineStage.RETRY_WAITING.value,
ChatGenerationPipelineStage.UPSCALE_QUEUED.value,
ChatGenerationPipelineStage.UPSCALE_PROCESSING.value,
ChatGenerationPipelineStage.UPSCALE_POLLING.value,
ChatGenerationPipelineStage.UPSCALE_DOWNLOADING.value,
ChatGenerationPipelineStage.UPSCALE_FINALIZING.value,
ChatGenerationPipelineStage.UPSCALE_RETRY_WAITING.value,
}
@@ -46,6 +53,8 @@ def get_display_status(task: ChatGenerationTask) -> str:
return "deleted"
if task.pipeline_stage == ChatGenerationPipelineStage.DOWNLOAD_FAILED.value:
return "download_failed"
if task.pipeline_stage == ChatGenerationPipelineStage.UPSCALE_FAILED.value:
return "failed"
return task.status or ChatGenerationTaskStatus.PENDING.value
@@ -115,6 +124,7 @@ def _generation_result_status(task: ChatGenerationTask) -> str:
if task.status == ChatGenerationTaskStatus.FAILED.value or (task.pipeline_stage or "") in {
ChatGenerationPipelineStage.FAILED.value,
ChatGenerationPipelineStage.TIMEOUT.value,
ChatGenerationPipelineStage.UPSCALE_FAILED.value,
}:
return "failed"
if is_task_active(task):
@@ -177,6 +187,7 @@ async def aggregate_main_task_status(
ChatGenerationPipelineStage.FAILED.value,
ChatGenerationPipelineStage.TIMEOUT.value,
ChatGenerationPipelineStage.DOWNLOAD_FAILED.value,
ChatGenerationPipelineStage.UPSCALE_FAILED.value,
}
)
]
@@ -191,11 +202,12 @@ async def aggregate_main_task_status(
main.error_message = _build_summary(children)
elif failed_children:
main.status = ChatGenerationTaskStatus.FAILED.value
main.pipeline_stage = (
ChatGenerationPipelineStage.DOWNLOAD_FAILED.value
if any(child.pipeline_stage == ChatGenerationPipelineStage.DOWNLOAD_FAILED.value for child in failed_children)
else ChatGenerationPipelineStage.FAILED.value
)
if any(child.pipeline_stage == ChatGenerationPipelineStage.DOWNLOAD_FAILED.value for child in failed_children):
main.pipeline_stage = ChatGenerationPipelineStage.DOWNLOAD_FAILED.value
elif any(child.pipeline_stage == ChatGenerationPipelineStage.UPSCALE_FAILED.value for child in failed_children):
main.pipeline_stage = ChatGenerationPipelineStage.UPSCALE_FAILED.value
else:
main.pipeline_stage = ChatGenerationPipelineStage.FAILED.value
main.generated_at = max(
(child.generated_at for child in completed_children if child.generated_at),
default=datetime.now(timezone.utc),
@@ -279,6 +291,7 @@ async def soft_delete_child_tasks_batch(
running_ids = [child.id for child in active_children if is_task_active(child)]
if running_ids:
raise HTTPException(status_code=400, detail=f"仍有 {len(running_ids)} 个子任务生成中，暂不能删除")
await assert_no_recoverable_failed_upscale_tasks(db, [str(child.id) for child in active_children])
if require_completed:
invalid_ids = [
child.id for child in active_children
@@ -381,6 +394,7 @@ async def soft_delete_top_level_task_group(
if task.generation_mode == GenerationMode.CHATAPI_ASYNC.value:
if is_task_active(task):
raise HTTPException(status_code=400, detail="当前任务正在生成中，暂不能删除")
await assert_no_recoverable_failed_upscale_tasks(db, [str(task.id)])
freed_size = await soft_delete_resources_by_source(
db,
source_model=SOURCE_MODEL_CHAT_TASK,
@@ -399,6 +413,7 @@ async def soft_delete_top_level_task_group(
active_children = [child for child in children if child.deleted_at is None]
child_ids = [child.id for child in active_children]
await assert_no_recoverable_failed_upscale_tasks(db, [str(child_id) for child_id in child_ids])
freed_size = await soft_delete_resources_by_source(
db,
source_model=SOURCE_MODEL_CHAT_TASK,
video-gen-api/app/services/generation/download_service.py
+40
View File
@@ -12,6 +12,7 @@ from app.services.provider_limit import provider_limit
from app.services.resource_accounting_service import safe_file_size
from app.services.video_cover_service import create_video_cover_for_local_video
from app.services.video_gen import download_video
from app.services.video_upscale.media_service import build_part_mp4_path, probe_video
@dataclass(slots=True)
@@ -98,6 +99,45 @@ async def _download_video_atomically(remote_url: str, final_path: str) -> str:
raise
async def download_video_upscale_source(record: ChatGenerationTask) -> DownloadedGenerationResult:
"""下载超分源视频。
源视频只供后处理使用，不生成封面，也不作为用户 GeneratedResource。
临时文件保持 .part.mp4 后缀，校验成功后原子重命名为 .source.mp4。
"""
if not record.remote_result_url:
raise ValueError("缺少远程结果URL")
date_dir = _build_storage_date_dir(record)
dest_dir = os.path.join(settings.STORAGE_LOCAL_PATH, "_upscale_source", date_dir)
os.makedirs(dest_dir, exist_ok=True)
dest = os.path.join(dest_dir, f"{record.id}.source.mp4")
if not _is_valid_file(dest):
part_path = build_part_mp4_path(dest)
try:
async with provider_limit("result_download", settings.RESULT_DOWNLOAD_MAX_CONCURRENCY):
await download_video(record.remote_result_url, part_path)
if not _is_valid_file(part_path):
raise RuntimeError("超分源视频下载完成但临时文件为空")
await probe_video(part_path)
os.replace(part_path, dest)
except Exception:
_safe_remove(part_path)
raise
else:
await probe_video(dest)
return DownloadedGenerationResult(
url=f"/generate/videos/_upscale_source/{date_dir}/{record.id}.source.mp4",
storage_path=dest,
file_size_bytes=safe_file_size(dest),
resource_type="video",
)
async def download_generation_result(record: ChatGenerationTask) -> DownloadedGenerationResult:
if not record.remote_result_url:
raise ValueError("缺少远程结果URL")
video-gen-api/app/services/generation/history_delete_service.py
+8
View File
@@ -28,6 +28,7 @@ from app.services.module_generation_flow_base_service import is_active_chat_gene
from app.services.module_generation_log_service import log_module_event_file
from app.services.operation_log_service import log_operation_event
# from app.services.operation_log import log_operation
from app.services.video_upscale.guard_service import assert_no_recoverable_failed_upscale_tasks
from app.services.resource_accounting_service import (
SOURCE_MODEL_CHAT_TASK,
SOURCE_MODEL_SHOT_SEGMENT,
@@ -204,6 +205,10 @@ async def _delete_generation_records(
)
records = list(result.scalars().all())
_raise_missing_if_any(ids=ids, found_ids=[record.id for record in records], message="项目生成记录不存在或已删除")
await assert_no_recoverable_failed_upscale_tasks(
db,
generation_record_ids=[str(record.id) for record in records],
)
invalid_ids = [
record.id
@@ -251,6 +256,8 @@ async def _delete_chat_tasks(
already_deleted_ids = [str(task.id) for task in tasks if task.deleted_at is not None]
_raise_invalid_if_any(invalid_ids=already_deleted_ids, message="AI 创作记录不存在或已删除", status_code=404)
await assert_no_recoverable_failed_upscale_tasks(db, [str(task.id) for task in tasks])
invalid_ids = [
task.id
for task in tasks
@@ -359,6 +366,7 @@ async def _soft_delete_module_projects(
_assert_no_active_chat_tasks(task_map.values())
chat_task_ids = list(task_map.keys())
await assert_no_recoverable_failed_upscale_tasks(db, chat_task_ids)
freed_size = await soft_delete_resources_by_source(
db,
source_model=SOURCE_MODEL_CHAT_TASK,
video-gen-api/app/services/generation/recovery_service.py
+6
View File
@@ -412,6 +412,12 @@ async def recover_one_generation_task(
await _remove_poll_active(task.id)
return "clean_not_generating"
if bool(getattr(task, "video_upscale_enabled_snapshot", False)) and str(task.pipeline_stage or "").startswith("upscale_"):
# 原视频已经进入超分流水线，后续由 video_upscale 恢复扫描处理。
# 这里禁止再次投递原结果下载，避免覆盖保留的 source.mp4 或提前生成用户资源。
await _remove_poll_active(task.id)
return "delegate_video_upscale_recovery"
has_remote_result = bool(str(task.remote_result_url or "").strip())
has_provider_task_id = bool(str(task.provider_task_id or "").strip() or str(task.seedance_task_id or "").strip())
is_deadline_expired = bool(task.deadline_at and _is_expired(task.deadline_at, current_time))
video-gen-api/app/services/generation/task_factory_service.py
+12
View File
@@ -27,6 +27,7 @@ from app.services.generation.ai.engine_service import (
)
from app.services.generation.billing_service import OWNER_CHAT_GENERATION_TASK, charge_generation_media_by_params
from app.services.resource_capacity_service import assert_user_resource_capacity_available
from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot
from app.services.private_portrait.reference_resolver import resolve_private_portrait_references
from app.utils.id_gen import generate_id
@@ -164,6 +165,12 @@ async def create_chat_generation_task_for_module(
raise HTTPException(status_code=400, detail=f"视频时长不支持: {selected_duration}")
if engine.max_duration and selected_duration > engine.max_duration:
raise HTTPException(status_code=400, detail=f"视频时长不能超过 {engine.max_duration} 秒")
provider_generation_resolution, upscale_enabled_snapshot, upscale_snapshot_json = await build_video_upscale_snapshot(
db,
target_resolution=selected_resolution,
aspect_ratio=ratio,
supported_provider_resolutions=resolutions,
)
media_billing = await charge_generation_media_by_params(
db,
user_id=current_user.id,
@@ -184,6 +191,8 @@ async def create_chat_generation_task_for_module(
)
snapshot = _build_video_snapshot(engine, ratio, selected_resolution, selected_duration)
snapshot["generation_count"] = 1
snapshot["provider_generation_resolution"] = provider_generation_resolution
snapshot["video_upscale_enabled_snapshot"] = upscale_enabled_snapshot
task = ChatGenerationTask(
id=task_id,
user_id=current_user.id,
@@ -193,6 +202,9 @@ async def create_chat_generation_task_for_module(
duration=selected_duration,
aspect_ratio=ratio,
resolution=selected_resolution,
provider_generation_resolution=provider_generation_resolution,
video_upscale_enabled_snapshot=upscale_enabled_snapshot,
video_upscale_snapshot_json=upscale_snapshot_json,
image_size=image_size or IMAGE_DEFAULT_SIZE,
image_proportion=image_proportion or IMAGE_DEFAULT_PROPORTION,
image_px=normalize_px(image_px) or IMAGE_DEFAULT_PX,
video-gen-api/app/services/module_generation_flow_base_service.py
+6
View File
@@ -242,6 +242,12 @@ ACTIVE_CHAT_TASK_BLOCK_STAGES = {
ChatGenerationPipelineStage.DOWNLOAD_QUEUED.value,
ChatGenerationPipelineStage.DOWNLOADING.value,
ChatGenerationPipelineStage.RETRY_WAITING.value,
ChatGenerationPipelineStage.UPSCALE_QUEUED.value,
ChatGenerationPipelineStage.UPSCALE_PROCESSING.value,
ChatGenerationPipelineStage.UPSCALE_POLLING.value,
ChatGenerationPipelineStage.UPSCALE_DOWNLOADING.value,
ChatGenerationPipelineStage.UPSCALE_FINALIZING.value,
ChatGenerationPipelineStage.UPSCALE_RETRY_WAITING.value,
}
video-gen-api/app/services/video_gen.py
+1 -1
View File
@@ -148,7 +148,7 @@ async def submit_video_task(
"content": content,
"ratio": record.aspect_ratio,
"duration": record.duration,
"resolution": record.resolution,
"resolution": getattr(record, "provider_generation_resolution", None) or record.resolution,
"generate_audio": True,
"watermark": False,
}
video-gen-api/app/services/video_queue.py
+200 -94
View File
@@ -2,123 +2,241 @@ import asyncio
import json
import logging
import os
from datetime import datetime
from datetime import datetime, timezone
from sqlalchemy import select
from sqlalchemy import or_, select
from app.config import settings
from app.enums.generation_status import GenerationRecordPipelineStage
from app.models.base import async_session
from app.models.generation_record import GenerationRecord
from app.services.video_gen import get_active_engine, poll_task_status, download_video, _log_video_response
from app.services.image_gen import get_active_image_engine, download_image
from app.services.generation.refund_service import mark_generation_record_failed_and_refund_once
from app.services.image_gen import download_image, get_active_image_engine
from app.services.media_token_usage_snapshot_service import sync_generation_record_media_token_snapshot
from app.services.provider_limit import provider_limit
from app.services.resource_accounting_service import (
record_generation_record_generated_resource,
safe_file_size,
)
from app.services.video_cover_service import create_video_cover_for_local_video
from app.config import settings
from app.services.generation.refund_service import mark_generation_record_failed_and_refund_once
from app.services.video_gen import _log_video_response, download_video, get_active_engine, poll_task_status
from app.services.video_upscale.media_service import build_part_mp4_path, probe_video, safe_remove
logger = logging.getLogger("videogen")
POLL_INTERVAL = 30 # seconds between polls
MAX_POLLS = 60 # max 30 minutes total
POLL_INTERVAL = 30
MAX_POLLS = 60
_PROVIDER_RECOVERABLE_STAGES = {
None,
"",
GenerationRecordPipelineStage.CREATING_PROVIDER_TASK.value,
GenerationRecordPipelineStage.WAITING_REMOTE.value,
GenerationRecordPipelineStage.POLLING.value,
GenerationRecordPipelineStage.RESULT_READY.value,
GenerationRecordPipelineStage.DOWNLOADING.value,
}
def _is_provider_stage(record: GenerationRecord) -> bool:
return (record.pipeline_stage or "") in _PROVIDER_RECOVERABLE_STAGES
def _source_date_dir(record: GenerationRecord) -> str:
created = record.created_at
if created is None:
created = datetime.now(timezone.utc)
return created.strftime("%Y/%m/%d")
async def _download_generation_record_upscale_source(
record: GenerationRecord,
remote_url: str,
) -> tuple[str, int]:
date_dir = _source_date_dir(record)
dest_dir = os.path.join(settings.STORAGE_LOCAL_PATH, "_upscale_source", date_dir)
os.makedirs(dest_dir, exist_ok=True)
final_path = os.path.join(dest_dir, f"{record.id}.source.mp4")
if os.path.isfile(final_path) and os.path.getsize(final_path) > 0:
await probe_video(final_path)
return final_path, safe_file_size(final_path)
part_path = build_part_mp4_path(final_path)
try:
async with provider_limit("result_download", settings.RESULT_DOWNLOAD_MAX_CONCURRENCY):
await download_video(remote_url, part_path)
if not os.path.isfile(part_path) or os.path.getsize(part_path) <= 0:
raise RuntimeError("超分源视频下载完成但临时文件为空")
await probe_video(part_path)
os.replace(part_path, final_path)
return final_path, safe_file_size(final_path)
except Exception:
safe_remove(part_path)
raise
async def handle_generation_record_video_succeeded(
db,
record: GenerationRecord,
*,
remote_url: str,
provider_response: dict | None,
video_tokens: int = 0,
) -> bool:
"""处理 GenerationRecord 原视频生成成功。
返回 True 表示已进入超分队列；False 表示按原流程直接完成。
"""
record.video_tokens_used = int(video_tokens or 0)
await sync_generation_record_media_token_snapshot(db, record, provider_response=provider_response or {})
if bool(record.video_upscale_enabled_snapshot) and record.video_upscale_snapshot_json:
record.pipeline_stage = GenerationRecordPipelineStage.DOWNLOADING.value
await db.flush()
source_path, source_size = await _download_generation_record_upscale_source(record, remote_url)
from app.services.video_upscale.task_service import enqueue_upscale_task, prepare_video_upscale_task
upscale = await prepare_video_upscale_task(
db,
generation_record=record,
source_local_path=source_path,
source_file_size_bytes=source_size,
source_remote_url=remote_url,
)
# enqueue_upscale_task 会先提交数据库，再投递 Celery；投递失败由 gen_recovery 补投。
await enqueue_upscale_task(db, upscale=upscale, reason="generation_record_source_ready")
logger.info("GenerationRecord 已进入视频超分队列: record_id=%s upscale_task_id=%s", record.id, upscale.id)
return True
storage_path = None
file_size_bytes = 0
if settings.STORAGE_TYPE == "local" and remote_url:
try:
date_dir = _source_date_dir(record)
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 = create_video_cover_for_local_video(
record_id=record.id,
video_path=dest,
date_dir=date_dir,
log_prefix=f"GenerationRecord视频封面生成 record_id={record.id}",
)
record.video_cover_url = cover_url
storage_path = dest
file_size_bytes = safe_file_size(dest)
except Exception as exc:
logger.warning("GenerationRecord 最终视频本地保存失败，回退远程地址: record_id=%s error=%s", record.id, exc)
record.video_url = remote_url
else:
record.video_url = remote_url
record.status = "completed"
record.pipeline_stage = GenerationRecordPipelineStage.DONE.value
record.generated_at = datetime.now(timezone.utc)
record.error_message = None
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,
)
await db.commit()
return False
class TaskQueue:
def __init__(self):
self.queue: asyncio.Queue[str] = asyncio.Queue()
self.running = False
self._active: dict[str, int] = {} # record_id -> poll count
self._active: dict[str, int] = {}
async def enqueue(self, record_id: str):
"""Add a record to the polling queue."""
await self.queue.put(record_id)
async def recover(self):
"""Recover in-progress tasks from DB on startup."""
async with async_session() as db:
result = await db.execute(
select(GenerationRecord).where(
GenerationRecord.status == "generating",
GenerationRecord.seedance_task_id.isnot(None),
GenerationRecord.deleted_at.is_(None),
or_(
GenerationRecord.pipeline_stage.is_(None),
GenerationRecord.pipeline_stage.in_(
[stage for stage in _PROVIDER_RECOVERABLE_STAGES if stage]
),
),
)
)
records = result.scalars().all()
for record in records:
await self.queue.put(record.id)
logger.info(f"Recovered task: {record.id} (seedance: {record.seedance_task_id})")
logger.info("Recovered task: %s (seedance: %s stage=%s)", record.id, record.seedance_task_id, record.pipeline_stage)
async def run(self):
"""Main polling loop."""
self.running = True
logger.info("Video queue started")
while self.running:
try:
record_id = await asyncio.wait_for(self.queue.get(), timeout=5.0)
except asyncio.TimeoutError:
continue
try:
await self._process(record_id)
except Exception as e:
logger.error(f"Error processing {record_id}: {e}")
except Exception as exc:
logger.exception("Error processing %s: %s", record_id, exc)
finally:
self.queue.task_done()
logger.info("Video queue stopped")
async def _process(self, record_id: str):
"""Process a single record: poll status and update DB."""
async with async_session() as db:
result = await db.execute(
select(GenerationRecord).where(
GenerationRecord.id == record_id,
GenerationRecord.deleted_at.is_(None),
)
.with_for_update()
.limit(1)
).with_for_update().limit(1)
)
record = result.scalar_one_or_none()
if not record or record.status != "generating":
return
if record.gen_type == "video":
if not _is_provider_stage(record):
return
await self._process_video(db, record)
else:
await self._process_image(db, record)
async def _process_video(self, db, record):
"""Process video generation task."""
async def _process_video(self, db, record: GenerationRecord):
record_id = record.id
if not record.seedance_task_id:
await mark_generation_record_failed_and_refund_once(
db,
record=record,
error_message="缺少外部任务ID",
)
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(db, record=record, error_message="缺少外部任务ID")
await db.commit()
return
record.pipeline_stage = GenerationRecordPipelineStage.POLLING.value
try:
engine = await get_active_engine(db)
poll_result = await poll_task_status(engine, record.seedance_task_id)
except Exception as e:
logger.error(f"Poll error for {record_id}: {e}")
except Exception as exc:
logger.error("Poll error for %s: %s", record_id, exc)
count = self._active.get(record_id, 0) + 1
self._active[record_id] = count
if count >= MAX_POLLS:
await mark_generation_record_failed_and_refund_once(
db,
record=record,
error_message=f"轮询超时: {e}",
)
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(db, record=record, error_message=f"轮询超时: {exc}")
await db.commit()
del self._active[record_id]
self._active.pop(record_id, None)
else:
await db.commit()
await asyncio.sleep(POLL_INTERVAL)
await self.queue.put(record_id)
return
@@ -128,54 +246,45 @@ class TaskQueue:
resp_data = json.loads(poll_result.get("response_data", "{}"))
except (json.JSONDecodeError, TypeError):
resp_data = {}
_log_video_response(record_id, resp_data, poll_result.get("error"))
if status == "succeeded":
file_url = poll_result.get("video_url", "")
storage_path = None
file_size_bytes = 0
if settings.STORAGE_TYPE == "local" and file_url:
try:
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(file_url, dest)
record.video_url = f"/generate/videos/{date_dir}/{record_id}.mp4"
cover_url, _cover_storage_path = create_video_cover_for_local_video(
record_id=record_id,
video_path=dest,
date_dir=date_dir,
log_prefix=f"GenerationRecord视频封面生成 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"Download failed, using remote URL: {e}")
record.video_url = file_url
else:
record.video_url = file_url
record.video_tokens_used = poll_result.get("video_tokens", 0)
await sync_generation_record_media_token_snapshot(db, record, provider_response=resp_data)
record.status = "completed"
record.generated_at = datetime.now()
if record.video_url:
await record_generation_record_generated_resource(
file_url = str(poll_result.get("video_url") or "").strip()
if not file_url:
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(db, record=record, error_message="供应商成功但未返回视频地址")
await db.commit()
return
record.pipeline_stage = GenerationRecordPipelineStage.RESULT_READY.value
await db.flush()
try:
await handle_generation_record_video_succeeded(
db,
record,
resource_url=record.video_url,
storage_path=storage_path,
file_size_bytes=file_size_bytes,
remote_url=file_url,
generated_at=record.generated_at,
provider_response=resp_data,
video_tokens=poll_result.get("video_tokens", 0),
)
except Exception as exc:
await db.rollback()
result = await db.execute(
select(GenerationRecord).where(GenerationRecord.id == record_id).with_for_update().limit(1)
)
failed_record = result.scalar_one_or_none()
if failed_record:
failed_record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(
db,
record=failed_record,
error_message=f"视频结果下载失败: {exc}",
)
await db.commit()
logger.exception("GenerationRecord 视频成功结果处理失败: %s", record_id)
self._active.pop(record_id, None)
await db.commit()
logger.info(f"Video task completed: {record_id}")
return
elif status == "failed":
if status == "failed":
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(
db,
record=record,
@@ -183,24 +292,21 @@ class TaskQueue:
)
self._active.pop(record_id, None)
await db.commit()
logger.info(f"Video task failed: {record_id}")
logger.info("Video task failed: %s", record_id)
return
count = self._active.get(record_id, 0) + 1
self._active[record_id] = count
if count >= MAX_POLLS:
record.pipeline_stage = GenerationRecordPipelineStage.FAILED.value
await mark_generation_record_failed_and_refund_once(db, record=record, error_message="视频生成超时")
self._active.pop(record_id, None)
await db.commit()
logger.info("Video task timed out: %s", record_id)
else:
count = self._active.get(record_id, 0) + 1
self._active[record_id] = count
if count >= MAX_POLLS:
await mark_generation_record_failed_and_refund_once(
db,
record=record,
error_message="视频生成超时",
)
self._active.pop(record_id, None)
await db.commit()
logger.info(f"Video task timed out: {record_id}")
else:
await db.commit()
await asyncio.sleep(POLL_INTERVAL)
await self.queue.put(record_id)
await db.commit()
await asyncio.sleep(POLL_INTERVAL)
await self.queue.put(record_id)
async def _process_image(self, db, record):
"""Process image generation task - calls API directly."""
video-gen-api/app/services/video_upscale/__init__.py
View File
video-gen-api/app/services/video_upscale/config_service.py
+175
View File
@@ -0,0 +1,175 @@
from __future__ import annotations
import json
from typing import Any
from fastapi import HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.video_upscale import (
ALL_PROCESSOR_KEYS,
VIDEO_UPSCALE_CONFIG_DESCRIPTION,
VIDEO_UPSCALE_CONFIG_KEY,
VIDEO_UPSCALE_CONFIG_VERSION,
VIDEO_UPSCALE_RESOLUTION_RANK,
VIDEO_UPSCALE_RESOLUTIONS,
VideoUpscaleProcessorKey,
normalize_video_upscale_resolution,
)
from app.models.system_config import SystemConfig
from app.schemas.video_upscale import VideoUpscaleConfigData
from app.utils.id_gen import generate_id
def default_video_upscale_config() -> dict[str, Any]:
return {
"enabled": False,
"version": VIDEO_UPSCALE_CONFIG_VERSION,
"delete_source_after_success": True,
"rules": [],
}
def _dump(data: dict[str, Any]) -> str:
return json.dumps(data, ensure_ascii=False, separators=(",", ":"), default=str)
def _load(value: str | None) -> dict[str, Any]:
if not value:
return default_video_upscale_config()
try:
data = json.loads(value)
except Exception as exc:
raise HTTPException(status_code=500, detail=f"视频超分配置 JSON 损坏: {exc}") from exc
if not isinstance(data, dict):
raise HTTPException(status_code=500, detail="视频超分配置必须是 JSON 对象")
return data
def _simplify_stored_config(data: dict[str, Any]) -> dict[str, Any]:
"""兼容开发阶段已保存的旧页面配置，忽略 processors、比例和手工像素等多余字段。"""
rules: list[dict[str, Any]] = []
for raw_rule in data.get("rules") or []:
if not isinstance(raw_rule, dict):
continue
rules.append(
{
"target_resolution": raw_rule.get("target_resolution"),
"provider_generation_resolution": raw_rule.get("provider_generation_resolution"),
"processor_key": raw_rule.get("processor_key"),
"enabled": bool(raw_rule.get("enabled", True)),
}
)
return {
"enabled": bool(data.get("enabled", False)),
"version": max(1, int(data.get("version") or VIDEO_UPSCALE_CONFIG_VERSION)),
"delete_source_after_success": bool(data.get("delete_source_after_success", True)),
"rules": rules,
}
async def _get_record(db: AsyncSession) -> SystemConfig | None:
result = await db.execute(select(SystemConfig).where(SystemConfig.key == VIDEO_UPSCALE_CONFIG_KEY).limit(1))
return result.scalar_one_or_none()
def validate_video_upscale_config(data: dict[str, Any] | VideoUpscaleConfigData) -> dict[str, Any]:
raw = data.model_dump() if isinstance(data, VideoUpscaleConfigData) else _simplify_stored_config(dict(data))
try:
model = VideoUpscaleConfigData.model_validate(raw)
except Exception as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
normalized = model.model_dump()
seen: set[str] = set()
for rule in normalized.get("rules") or []:
target = normalize_video_upscale_resolution(rule.get("target_resolution"))
provider_resolution = normalize_video_upscale_resolution(rule.get("provider_generation_resolution"))
processor_key = str(rule.get("processor_key") or "").strip()
if target not in VIDEO_UPSCALE_RESOLUTIONS:
raise HTTPException(
status_code=400,
detail=f"不支持的客户目标分辨率: {target}，仅支持 {'、'.join(VIDEO_UPSCALE_RESOLUTIONS)}",
)
if provider_resolution not in VIDEO_UPSCALE_RESOLUTIONS:
raise HTTPException(
status_code=400,
detail=f"不支持的实际生成分辨率: {provider_resolution}，仅支持 {'、'.join(VIDEO_UPSCALE_RESOLUTIONS)}",
)
if processor_key not in ALL_PROCESSOR_KEYS:
raise HTTPException(status_code=400, detail=f"未注册的超分处理器: {processor_key}")
rule["target_resolution"] = target
rule["provider_generation_resolution"] = provider_resolution
rule["processor_key"] = processor_key
if VIDEO_UPSCALE_RESOLUTION_RANK[provider_resolution] > VIDEO_UPSCALE_RESOLUTION_RANK[target]:
raise HTTPException(
status_code=400,
detail=f"规则 {target} 的实际生成分辨率 {provider_resolution} 不能高于客户目标分辨率",
)
if processor_key == VideoUpscaleProcessorKey.VOLC_LARGE_MODEL_V1.value:
if target not in {"720p", "1080p", "2K"}:
raise HTTPException(status_code=400, detail="火山画质增强大模型目标分辨率仅支持 720p、1080p、2K")
if VIDEO_UPSCALE_RESOLUTION_RANK[provider_resolution] > VIDEO_UPSCALE_RESOLUTION_RANK["1080p"]:
raise HTTPException(status_code=400, detail="火山画质增强大模型输入视频最高支持 1080p")
if processor_key in {
VideoUpscaleProcessorKey.VOLC_STANDARD_V1.value,
VideoUpscaleProcessorKey.VOLC_PROFESSIONAL_V1.value,
} and VIDEO_UPSCALE_RESOLUTION_RANK[provider_resolution] > VIDEO_UPSCALE_RESOLUTION_RANK["2K"]:
raise HTTPException(status_code=400, detail="火山标准版/专业版输入视频最高支持 2K")
if rule.get("enabled"):
if target in seen:
raise HTTPException(status_code=400, detail=f"客户目标分辨率存在重复启用规则: {target}")
seen.add(target)
return normalized
async def get_video_upscale_config(db: AsyncSession) -> dict[str, Any]:
record = await _get_record(db)
data = default_video_upscale_config() if record is None else validate_video_upscale_config(_load(record.value))
return {
"id": record.id if record else None,
"key": VIDEO_UPSCALE_CONFIG_KEY,
"description": record.description if record else VIDEO_UPSCALE_CONFIG_DESCRIPTION,
"data": data,
"created_at": record.created_at if record else None,
"updated_at": record.updated_at if record else None,
}
async def get_runtime_video_upscale_config(db: AsyncSession) -> dict[str, Any]:
record = await _get_record(db)
if record is None:
return default_video_upscale_config()
return validate_video_upscale_config(_load(record.value))
async def save_video_upscale_config(db: AsyncSession, data: dict[str, Any] | VideoUpscaleConfigData) -> dict[str, Any]:
normalized = validate_video_upscale_config(data)
record = await _get_record(db)
old_version = 0
if record:
try:
old_version = int((_load(record.value).get("version") or 0))
except Exception:
old_version = 0
normalized["version"] = max(old_version + 1, VIDEO_UPSCALE_CONFIG_VERSION)
if record is None:
record = SystemConfig(
id=generate_id(),
key=VIDEO_UPSCALE_CONFIG_KEY,
value=_dump(normalized),
description=VIDEO_UPSCALE_CONFIG_DESCRIPTION,
)
db.add(record)
else:
record.value = _dump(normalized)
record.description = VIDEO_UPSCALE_CONFIG_DESCRIPTION
await db.flush()
return await get_video_upscale_config(db)
video-gen-api/app/services/video_upscale/guard_service.py
+61
View File
@@ -0,0 +1,61 @@
from __future__ import annotations
import os
from collections.abc import Iterable
from fastapi import HTTPException
from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.video_upscale import VideoUpscaleTaskStatus
from app.models.video_upscale_task import VideoUpscaleTask
async def assert_no_recoverable_failed_upscale_tasks(
db: AsyncSession,
chat_task_ids: Iterable[str] | None = None,
*,
generation_record_ids: Iterable[str] | None = None,
) -> None:
"""阻止删除仍保留本地源视频、可由管理员人工恢复的超分失败任务。"""
chat_ids = list(dict.fromkeys(str(item) for item in (chat_task_ids or []) if item))
record_ids = list(dict.fromkeys(str(item) for item in (generation_record_ids or []) if item))
if not chat_ids and not record_ids:
return
owner_conditions = []
if chat_ids:
owner_conditions.append(VideoUpscaleTask.chat_generation_task_id.in_(chat_ids))
if record_ids:
owner_conditions.append(VideoUpscaleTask.generation_record_id.in_(record_ids))
result = await db.execute(
select(
VideoUpscaleTask.chat_generation_task_id,
VideoUpscaleTask.generation_record_id,
VideoUpscaleTask.source_local_path,
).where(
or_(*owner_conditions),
VideoUpscaleTask.status == VideoUpscaleTaskStatus.FAILED.value,
VideoUpscaleTask.source_local_path.isnot(None),
VideoUpscaleTask.source_deleted_at.is_(None),
)
)
recoverable = []
for chat_task_id, generation_record_id, source_local_path in result.all():
if source_local_path and os.path.isfile(str(source_local_path)):
recoverable.append(
{
"owner_type": "chat_generation_task" if chat_task_id else "generation_record",
"owner_id": str(chat_task_id or generation_record_id),
}
)
if recoverable:
raise HTTPException(
status_code=409,
detail={
"message": "当前生成任务异常暂不能删除",
"items": recoverable,
"task_count": len(recoverable),
},
)
video-gen-api/app/services/video_upscale/local_ffmpeg_service.py
+128
View File
@@ -0,0 +1,128 @@
from __future__ import annotations
import asyncio
import os
import subprocess
from pathlib import Path
from app.config import settings
from app.services.video_cover_service import get_ffmpeg_bin
from app.services.video_upscale.media_service import build_part_mp4_path, is_valid_file, probe_video, safe_remove
class LocalVideoUpscaleError(RuntimeError):
pass
def _build_filter(target_width: int, target_height: int) -> str:
if target_width <= 0 or target_height <= 0 or target_width % 2 or target_height % 2:
raise LocalVideoUpscaleError("目标宽高必须是正偶数")
return (
"hqdn3d=0.8:0.6:2.5:1.8,"
f"scale={target_width}:{target_height}:"
"force_original_aspect_ratio=increase:force_divisible_by=2:flags=lanczos,"
f"crop={target_width}:{target_height}:(iw-ow)/2:(ih-oh)/2,"
"unsharp=5:5:0.40:5:5:0.0,"
"eq=contrast=1.02:saturation=1.03,"
"setsar=1"
)
def _run_ffmpeg_sync(
*,
source_path: str,
part_path: str,
target_width: int,
target_height: int,
timeout_seconds: int,
) -> None:
ffmpeg_bin = get_ffmpeg_bin() # 明确复用 config.py 的 FFMPEG_BIN。
cmd = [
ffmpeg_bin,
"-hide_banner",
"-nostdin",
"-y",
"-i",
source_path,
"-map",
"0:v:0",
"-map",
"0:a?",
"-vf",
_build_filter(target_width, target_height),
"-c:v",
"libx264",
"-preset",
"slow",
"-crf",
"18",
"-profile:v",
"high",
"-pix_fmt",
"yuv420p",
"-movflags",
"+faststart",
"-c:a",
"aac",
"-b:a",
"192k",
part_path,
]
try:
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
timeout=max(30, int(timeout_seconds)),
shell=False,
)
except subprocess.TimeoutExpired as exc:
raise LocalVideoUpscaleError(f"本地 FFmpeg 超分超时: {timeout_seconds} 秒") from exc
except OSError as exc:
raise LocalVideoUpscaleError(f"本地 FFmpeg 无法启动: {exc}") from exc
if result.returncode != 0:
raise LocalVideoUpscaleError(f"本地 FFmpeg 超分失败: {(result.stderr or '').strip()[-4000:]}")
async def execute_local_ffmpeg_crop(
*,
source_path: str,
final_path: str,
target_width: int,
target_height: int,
timeout_seconds: int | None = None,
) -> str:
if not is_valid_file(source_path):
raise LocalVideoUpscaleError(f"超分源视频不存在或为空: {source_path}")
if is_valid_file(final_path):
info = await probe_video(final_path)
if info.width == target_width and info.height == target_height:
return final_path
os.makedirs(os.path.dirname(final_path), exist_ok=True)
part_path = build_part_mp4_path(final_path)
safe_remove(part_path)
try:
await asyncio.to_thread(
_run_ffmpeg_sync,
source_path=source_path,
part_path=part_path,
target_width=target_width,
target_height=target_height,
timeout_seconds=int(timeout_seconds or settings.VIDEO_UPSCALE_LOCAL_TIMEOUT_SECONDS),
)
if not is_valid_file(part_path):
raise LocalVideoUpscaleError("本地 FFmpeg 输出文件为空")
info = await probe_video(part_path)
if info.width != target_width or info.height != target_height:
raise LocalVideoUpscaleError(
f"本地 FFmpeg 输出尺寸不正确: {info.width}x{info.height}，预期 {target_width}x{target_height}"
)
os.replace(part_path, final_path)
return final_path
except Exception:
safe_remove(part_path)
raise
video-gen-api/app/services/video_upscale/log_service.py
+108
View File
@@ -0,0 +1,108 @@
from __future__ import annotations
from typing import Any
from urllib.parse import urlsplit, urlunsplit
from app.models.chat_generation_task import ChatGenerationTask
from app.models.generation_record import GenerationRecord
from app.services.operation_log_service import log_operation_event
def _sanitize_url(value: Any) -> Any:
if not isinstance(value, str) or not value.startswith(("http://", "https://")):
return value
parts = urlsplit(value)
return urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))
def _sanitize_detail(value: Any) -> Any:
if isinstance(value, dict):
result: dict[str, Any] = {}
for key, item in value.items():
key_text = str(key).lower()
if key_text in {"authorization", "api_key", "volc_api_key"}:
result[key] = "***"
elif "url" in key_text:
result[key] = _sanitize_url(item)
else:
result[key] = _sanitize_detail(item)
return result
if isinstance(value, list):
return [_sanitize_detail(item) for item in value]
return value
def _owner_context(task: Any | None, upscale_task: Any | None) -> dict[str, Any]:
if isinstance(task, ChatGenerationTask):
return {
"owner_type": "chat_generation_task",
"owner_id": task.id,
"chat_generation_task_id": task.id,
"generation_record_id": None,
"project_id": None,
"generation_mode": task.generation_mode,
}
if isinstance(task, GenerationRecord):
return {
"owner_type": "generation_record",
"owner_id": task.id,
"chat_generation_task_id": None,
"generation_record_id": task.id,
"project_id": task.project_id,
"generation_mode": None,
}
chat_task_id = getattr(upscale_task, "chat_generation_task_id", None) if upscale_task else None
generation_record_id = getattr(upscale_task, "generation_record_id", None) if upscale_task else None
return {
"owner_type": "chat_generation_task" if chat_task_id else "generation_record" if generation_record_id else None,
"owner_id": chat_task_id or generation_record_id,
"chat_generation_task_id": chat_task_id,
"generation_record_id": generation_record_id,
"project_id": None,
"generation_mode": None,
}
def log_video_upscale_event(
*,
event_type: str,
event_status: str = "success",
task: Any | None = None,
upscale_task: Any | None = None,
remote_request_id: str | None = None,
message: str | None = None,
detail: dict[str, Any] | None = None,
error: str | None = None,
) -> None:
"""统一复用 operation_log_service 写入视频超分步骤日志。"""
final_detail = _owner_context(task, upscale_task)
final_detail.update(_sanitize_detail(dict(detail or {})))
if upscale_task is not None:
final_detail.update(
{
"upscale_task_id": getattr(upscale_task, "id", None),
"processor_key": getattr(upscale_task, "processor_key", None),
"upscale_status": getattr(upscale_task, "status", None),
"upscale_stage": getattr(upscale_task, "stage", None),
"attempt_count": getattr(upscale_task, "attempt_count", None),
"failure_count": getattr(upscale_task, "failure_count", None),
"provider_task_id": getattr(upscale_task, "provider_task_id", None),
"input_source_type": getattr(upscale_task, "input_source_type", None),
"target_width": getattr(upscale_task, "target_width", None),
"target_height": getattr(upscale_task, "target_height", None),
}
)
owner_id = final_detail.get("owner_id")
log_operation_event(
domain="video_upscale",
event_type=event_type,
event_status=event_status,
source="service",
user_id=getattr(task, "user_id", None) if task is not None else None,
group_id=getattr(task, "parent_task_id", None) if task is not None else None,
task_id=owner_id,
remote_request_id=remote_request_id,
message=message,
detail=final_detail,
error=error,
)
video-gen-api/app/services/video_upscale/media_service.py
+207
View File
@@ -0,0 +1,207 @@
from __future__ import annotations
import asyncio
import json
import os
import shutil
import subprocess
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from urllib.parse import parse_qsl, urlsplit
import httpx
from app.config import settings
from app.services.resource_signed_url_service import build_resource_signed_url
from app.services.video_cover_service import get_ffmpeg_bin
@dataclass(slots=True)
class VideoMediaInfo:
width: int
height: int
duration_seconds: float
fps: float
codec_name: str | None = None
color_transfer: str | None = None
color_primaries: str | None = None
color_space: str | None = None
def build_part_mp4_path(final_path: str) -> str:
path = Path(final_path)
return str(path.with_name(f"{path.stem}.{uuid.uuid4().hex}.part.mp4"))
def safe_remove(path: str | None) -> bool:
if not path:
return True
try:
if os.path.exists(path):
os.remove(path)
return not os.path.exists(path)
except OSError:
return False
def is_valid_file(path: str | None) -> bool:
if not path:
return False
try:
return os.path.isfile(path) and os.path.getsize(path) > 0
except OSError:
return False
def get_ffprobe_bin() -> str:
ffmpeg = Path(get_ffmpeg_bin())
sibling = ffmpeg.with_name("ffprobe.exe" if ffmpeg.suffix.lower() == ".exe" else "ffprobe")
if sibling.exists():
return str(sibling)
found = shutil.which("ffprobe") or shutil.which("ffprobe.exe")
if found:
return found
raise RuntimeError("未找到 ffprobe，请确保其与 FFMPEG_BIN 同目录或已加入 PATH")
def _parse_fps(value: str | None) -> float:
if not value:
return 0.0
try:
if "/" in value:
left, right = value.split("/", 1)
denominator = float(right)
return float(left) / denominator if denominator else 0.0
return float(value)
except Exception:
return 0.0
def probe_video_sync(path: str, timeout_seconds: int = 30) -> VideoMediaInfo:
if not is_valid_file(path):
raise RuntimeError(f"视频文件不存在或为空: {path}")
cmd = [
get_ffprobe_bin(), "-v", "error", "-select_streams", "v:0",
"-show_entries", "stream=width,height,codec_name,avg_frame_rate,color_transfer,color_primaries,color_space:format=duration",
"-of", "json", path,
]
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
timeout=max(5, int(timeout_seconds)),
shell=False,
)
if result.returncode != 0:
raise RuntimeError(f"ffprobe 校验失败: {(result.stderr or '').strip()[-2000:]}")
try:
payload = json.loads(result.stdout or "{}")
stream = (payload.get("streams") or [])[0]
width = int(stream.get("width") or 0)
height = int(stream.get("height") or 0)
duration = float((payload.get("format") or {}).get("duration") or 0)
except Exception as exc:
raise RuntimeError(f"ffprobe 响应解析失败: {exc}") from exc
if width <= 0 or height <= 0:
raise RuntimeError("ffprobe 未读取到有效视频宽高")
return VideoMediaInfo(
width=width,
height=height,
duration_seconds=duration,
fps=_parse_fps(stream.get("avg_frame_rate")),
codec_name=stream.get("codec_name"),
color_transfer=stream.get("color_transfer"),
color_primaries=stream.get("color_primaries"),
color_space=stream.get("color_space"),
)
async def probe_video(path: str, timeout_seconds: int = 30) -> VideoMediaInfo:
return await asyncio.to_thread(probe_video_sync, path, timeout_seconds)
def parse_tos_signed_url_expiry(url: str | None) -> tuple[datetime | None, datetime | None]:
if not url:
return None, None
params = {key.lower(): value for key, value in parse_qsl(urlsplit(url).query, keep_blank_values=True)}
date_text = params.get("x-tos-date")
expires_text = params.get("x-tos-expires")
if not date_text or not expires_text:
return None, None
try:
signed_at = datetime.strptime(date_text, "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc)
expires_seconds = int(expires_text)
if expires_seconds <= 0:
return None, None
return signed_at, signed_at + timedelta(seconds=expires_seconds)
except Exception:
return None, None
async def probe_remote_url(url: str) -> bool:
timeout = httpx.Timeout(
connect=max(1, int(settings.VIDEO_UPSCALE_REMOTE_URL_PROBE_CONNECT_TIMEOUT_SECONDS or 3)),
read=max(1, int(settings.VIDEO_UPSCALE_REMOTE_URL_PROBE_READ_TIMEOUT_SECONDS or 5)),
write=5,
pool=5,
)
try:
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
async with client.stream("GET", url, headers={"Range": "bytes=0-0"}) as response:
if response.status_code not in {200, 206}:
return False
async for chunk in response.aiter_bytes():
return bool(chunk)
return False
except Exception:
return False
def build_source_resource_url(source_local_path: str) -> str:
root = Path(settings.STORAGE_LOCAL_PATH).resolve()
path = Path(source_local_path).resolve()
try:
relative = path.relative_to(root).as_posix()
except ValueError as exc:
raise RuntimeError("超分源视频不在 STORAGE_LOCAL_PATH 下，无法生成签名 URL") from exc
return f"{settings.BASE_URL.rstrip('/')}/generate/videos/{relative}"
def build_local_source_signed_url(source_local_path: str, expire_seconds: int) -> str:
return build_resource_signed_url(
build_source_resource_url(source_local_path),
expire_seconds=max(600, int(expire_seconds)),
)
async def download_video_to_path(url: str, final_path: str, timeout_seconds: int) -> str:
if is_valid_file(final_path):
try:
await probe_video(final_path)
return final_path
except Exception:
safe_remove(final_path)
os.makedirs(os.path.dirname(final_path), exist_ok=True)
part_path = build_part_mp4_path(final_path)
timeout = httpx.Timeout(max(30, int(timeout_seconds)))
try:
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
async with client.stream("GET", url) as response:
response.raise_for_status()
with open(part_path, "wb") as file_obj:
async for chunk in response.aiter_bytes(chunk_size=1024 * 1024):
file_obj.write(chunk)
if not is_valid_file(part_path):
raise RuntimeError("视频下载完成但临时文件为空")
await probe_video(part_path)
os.replace(part_path, final_path)
return final_path
except Exception:
safe_remove(part_path)
raise
video-gen-api/app/services/video_upscale/owner_service.py
+120
View File
@@ -0,0 +1,120 @@
from __future__ import annotations
from typing import Any, TypeAlias
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.generation_status import GenerationRecordPipelineStage, GenerationStatus
from app.enums.generation_task import ChatGenerationPipelineStage, ChatGenerationTaskStatus
from app.models.chat_generation_task import ChatGenerationTask
from app.models.generation_record import GenerationRecord
from app.models.video_upscale_task import VideoUpscaleTask
from app.services.generation.ai.task_group_service import aggregate_parent_for_child
from app.services.generation.module_hook_service import notify_chat_generation_task_finished
VideoUpscaleOwner: TypeAlias = ChatGenerationTask | GenerationRecord
def owner_type(owner: VideoUpscaleOwner | None) -> str | None:
if isinstance(owner, ChatGenerationTask):
return "chat_generation_task"
if isinstance(owner, GenerationRecord):
return "generation_record"
return None
def owner_id(owner: VideoUpscaleOwner | None) -> str | None:
return str(getattr(owner, "id", "") or "") or None
def owner_is_generating(owner: VideoUpscaleOwner) -> bool:
if isinstance(owner, ChatGenerationTask):
return owner.status == ChatGenerationTaskStatus.GENERATING.value
return owner.status == GenerationStatus.generating.value
def owner_is_completed(owner: VideoUpscaleOwner) -> bool:
if isinstance(owner, ChatGenerationTask):
return owner.status == ChatGenerationTaskStatus.COMPLETED.value
return owner.status == GenerationStatus.completed.value
def set_owner_stage(owner: VideoUpscaleOwner, stage: str) -> None:
owner.pipeline_stage = stage
def upscale_stage_value(owner: VideoUpscaleOwner, chat_stage: ChatGenerationPipelineStage | str) -> str:
value = chat_stage.value if hasattr(chat_stage, "value") else str(chat_stage)
if isinstance(owner, ChatGenerationTask):
return value
try:
return GenerationRecordPipelineStage(value).value
except ValueError:
return value
async def load_upscale_owner(
db: AsyncSession,
upscale: VideoUpscaleTask,
*,
for_update: bool,
) -> VideoUpscaleOwner | None:
if upscale.chat_generation_task_id:
query = select(ChatGenerationTask).where(
ChatGenerationTask.id == upscale.chat_generation_task_id,
ChatGenerationTask.deleted_at.is_(None),
)
elif upscale.generation_record_id:
query = select(GenerationRecord).where(
GenerationRecord.id == upscale.generation_record_id,
GenerationRecord.deleted_at.is_(None),
)
else:
return None
if for_update:
query = query.with_for_update()
result = await db.execute(query.limit(1))
return result.scalar_one_or_none()
async def mark_owner_upscale_failed(
db: AsyncSession,
owner: VideoUpscaleOwner,
*,
error_message: str,
) -> None:
if isinstance(owner, ChatGenerationTask):
owner.status = ChatGenerationTaskStatus.FAILED.value
owner.pipeline_stage = ChatGenerationPipelineStage.UPSCALE_FAILED.value
owner.error_message = error_message
await notify_chat_generation_task_finished(db, owner)
await aggregate_parent_for_child(db, owner)
return
owner.status = GenerationStatus.failed.value
owner.pipeline_stage = GenerationRecordPipelineStage.UPSCALE_FAILED.value
owner.error_message = error_message
def restore_owner_for_upscale_retry(owner: VideoUpscaleOwner) -> None:
if isinstance(owner, ChatGenerationTask):
owner.status = ChatGenerationTaskStatus.GENERATING.value
owner.pipeline_stage = ChatGenerationPipelineStage.UPSCALE_QUEUED.value
else:
owner.status = GenerationStatus.generating.value
owner.pipeline_stage = GenerationRecordPipelineStage.UPSCALE_QUEUED.value
owner.error_message = None
def owner_context(owner: VideoUpscaleOwner | None) -> dict[str, Any]:
if owner is None:
return {}
return {
"owner_type": owner_type(owner),
"owner_id": owner_id(owner),
"chat_generation_task_id": owner.id if isinstance(owner, ChatGenerationTask) else None,
"generation_record_id": owner.id if isinstance(owner, GenerationRecord) else None,
"project_id": getattr(owner, "project_id", None),
"generation_mode": getattr(owner, "generation_mode", None),
}
video-gen-api/app/services/video_upscale/snapshot_service.py
+176
View File
@@ -0,0 +1,176 @@
from __future__ import annotations
import hashlib
import json
from datetime import datetime, timezone
from typing import Any, Iterable
from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.enums.video_upscale import (
ALL_PROCESSOR_KEYS,
VIDEO_UPSCALE_RESOLUTIONS,
VideoUpscaleProcessorKey,
normalize_video_upscale_resolution,
video_upscale_short_edge_pixels,
)
from app.services.video_upscale.config_service import get_runtime_video_upscale_config
from app.services.video_upscale.log_service import log_video_upscale_event
def normalize_resolution(value: str | None) -> str:
return normalize_video_upscale_resolution(value)
def _even(value: float) -> int:
rounded = int(round(value))
if rounded < 2:
rounded = 2
return rounded if rounded % 2 == 0 else rounded + 1
def parse_aspect_ratio(value: str | None) -> tuple[int, int]:
text = str(value or "").strip()
try:
left, right = text.split(":", 1)
width = int(left)
height = int(right)
except Exception as exc:
raise HTTPException(status_code=400, detail=f"视频比例格式不合法: {text}") from exc
if width <= 0 or height <= 0:
raise HTTPException(status_code=400, detail=f"视频比例格式不合法: {text}")
return width, height
def calculate_target_dimensions(*, aspect_ratio: str, target_resolution: str) -> tuple[int, int]:
ratio_w, ratio_h = parse_aspect_ratio(aspect_ratio)
try:
pixels = video_upscale_short_edge_pixels(target_resolution)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
if ratio_w >= ratio_h:
height = _even(pixels)
width = _even(height * ratio_w / ratio_h)
else:
width = _even(pixels)
height = _even(width * ratio_h / ratio_w)
return width, height
def _runtime_processor_snapshot(processor_key: str) -> dict[str, Any]:
if processor_key not in ALL_PROCESSOR_KEYS:
raise HTTPException(status_code=500, detail=f"未注册的超分处理器: {processor_key}")
is_local = processor_key == VideoUpscaleProcessorKey.LOCAL_FFMPEG_CROP_V1.value
return {
"max_attempts": max(1, int(settings.VIDEO_UPSCALE_MAX_ATTEMPTS or 3)),
"timeout_seconds": int(
settings.VIDEO_UPSCALE_LOCAL_TIMEOUT_SECONDS
if is_local
else settings.VIDEO_UPSCALE_REMOTE_POLL_TIMEOUT_SECONDS
),
"request_timeout_seconds": max(3, int(settings.VIDEO_UPSCALE_REMOTE_REQUEST_TIMEOUT_SECONDS or 30)),
"poll_timeout_seconds": max(60, int(settings.VIDEO_UPSCALE_REMOTE_POLL_TIMEOUT_SECONDS or 7200)),
"poll_interval_seconds": max(5, int(settings.VIDEO_UPSCALE_REMOTE_POLL_INTERVAL_SECONDS or 30)),
"source_url_expire_seconds": max(600, int(settings.VIDEO_UPSCALE_LOCAL_SIGNED_URL_EXPIRE_SECONDS or 7200)),
"bitrate_level": "medium",
"scene": "aigc",
"fps": None,
"queue_id": None,
}
async def build_video_upscale_snapshot(
db: AsyncSession,
*,
target_resolution: str,
aspect_ratio: str,
supported_provider_resolutions: Iterable[str] | None = None,
) -> tuple[str, bool, str | None]:
config = await get_runtime_video_upscale_config(db)
original_resolution = normalize_resolution(target_resolution)
if original_resolution not in VIDEO_UPSCALE_RESOLUTIONS:
return str(target_resolution).strip(), False, None
if not config.get("enabled"):
log_video_upscale_event(
event_type="upscale_snapshot_bypassed",
event_status="bypassed",
detail={"reason": "global_disabled", "target_resolution": original_resolution, "aspect_ratio": aspect_ratio},
)
return original_resolution, False, None
matched: dict[str, Any] | None = None
for rule in config.get("rules") or []:
if not rule.get("enabled"):
continue
if normalize_resolution(rule.get("target_resolution")) == original_resolution:
matched = dict(rule)
break
if matched is None:
log_video_upscale_event(
event_type="upscale_snapshot_bypassed",
event_status="bypassed",
detail={"reason": "rule_not_matched", "target_resolution": original_resolution, "aspect_ratio": aspect_ratio},
)
return original_resolution, False, None
provider_resolution = normalize_resolution(matched.get("provider_generation_resolution"))
processor_key = str(matched.get("processor_key") or "").strip()
processor = _runtime_processor_snapshot(processor_key)
supported = {normalize_resolution(item) for item in (supported_provider_resolutions or []) if item}
if supported and provider_resolution not in supported:
raise HTTPException(
status_code=400,
detail=f"超分规则要求实际生成 {provider_resolution}，但当前视频引擎不支持该分辨率",
)
target_width, target_height = calculate_target_dimensions(
aspect_ratio=aspect_ratio,
target_resolution=original_resolution,
)
target_short_edge_pixels = video_upscale_short_edge_pixels(original_resolution)
snapshot: dict[str, Any] = {
"config_version": int(config.get("version") or 1),
"target_resolution": original_resolution,
"provider_generation_resolution": provider_resolution,
"processor_key": processor_key,
"processor": processor,
"aspect_ratio": aspect_ratio,
"target_short_edge_pixels": target_short_edge_pixels,
"target_width": target_width,
"target_height": target_height,
"delete_source_after_success": bool(config.get("delete_source_after_success", True)),
"snapshot_created_at": datetime.now(timezone.utc).isoformat(),
}
canonical = json.dumps(snapshot, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
snapshot["snapshot_hash"] = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
log_video_upscale_event(
event_type="upscale_snapshot_matched",
detail={
"target_resolution": original_resolution,
"provider_generation_resolution": provider_resolution,
"processor_key": processor_key,
"aspect_ratio": aspect_ratio,
"target_width": target_width,
"target_height": target_height,
"delete_source_after_success": snapshot["delete_source_after_success"],
"config_version": snapshot["config_version"],
"snapshot_hash": snapshot["snapshot_hash"],
},
)
return provider_resolution, True, json.dumps(snapshot, ensure_ascii=False, separators=(",", ":"))
def parse_video_upscale_snapshot(value: str | None) -> dict[str, Any]:
if not value:
return {}
try:
data = json.loads(value)
except Exception as exc:
raise RuntimeError(f"超分快照不是合法 JSON: {exc}") from exc
if not isinstance(data, dict):
raise RuntimeError("超分快照必须是 JSON 对象")
return data
video-gen-api/app/services/video_upscale/task_service.py
+1431
View File
File diff suppressed because it is too large Load Diff
video-gen-api/app/services/video_upscale/volc_service.py
+270
View File
@@ -0,0 +1,270 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
import httpx
from app.config import settings
from app.enums.video_upscale import VideoUpscaleProcessorKey
class VolcMediaKitError(RuntimeError):
def __init__(
self,
message: str,
*,
code: str | None = None,
error_type: str | None = None,
param: str | None = None,
retryable: bool = True,
http_status: int | None = None,
request_id: str | None = None,
endpoint: str | None = None,
response_payload: dict[str, Any] | None = None,
):
super().__init__(message)
self.code = code
self.error_type = error_type
self.param = param
self.retryable = retryable
self.http_status = http_status
self.request_id = request_id
self.endpoint = endpoint
self.response_payload = response_payload or {}
def log_detail(self) -> dict[str, Any]:
return {
"endpoint": self.endpoint,
"http_status": self.http_status,
"request_id": self.request_id,
"error_code": self.code,
"error_type": self.error_type,
"error_param": self.param,
"error_message": str(self),
"retryable": self.retryable,
}
@dataclass(slots=True)
class VolcSubmitResult:
task_id: str
request_id: str | None
request_payload: dict[str, Any]
response_payload: dict[str, Any]
endpoint: str
@dataclass(slots=True)
class VolcQueryResult:
status: str
request_id: str | None
result: dict[str, Any] | None
error: dict[str, Any] | None
expires_at: int | None
response_payload: dict[str, Any]
endpoint: str
def _headers() -> dict[str, str]:
api_key = str(settings.VOLC_API_KEY or "").strip()
if not api_key:
raise VolcMediaKitError("VOLC_API_KEY 未配置", code="MissingApiKey", retryable=False)
return {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
def _base_url() -> str:
return str(settings.VOLC_MEDIAKIT_API_BASE or "https://mediakit.cn-beijing.volces.com").rstrip("/")
def _error_from_payload(
payload: dict[str, Any],
default_message: str,
*,
http_status: int | None = None,
endpoint: str | None = None,
) -> VolcMediaKitError:
error = payload.get("error") if isinstance(payload.get("error"), dict) else {}
code = str(error.get("code") or "") or None
error_type = str(error.get("type") or "") or None
param = str(error.get("param") or "") or None
message = str(error.get("message") or default_message)
retryable = True
if http_status in {400, 401, 403, 404, 422}:
retryable = False
if code in {"InvalidParameter", "Unauthorized", "Forbidden", "NotFound"} or error_type in {"BadRequest", "AuthError"}:
retryable = False
request_id = str(payload.get("request_id") or "") or None
return VolcMediaKitError(
message,
code=code,
error_type=error_type,
param=param,
retryable=retryable,
http_status=http_status,
request_id=request_id,
endpoint=endpoint,
response_payload=payload,
)
def build_submit_payload(
*,
processor_key: str,
video_url: str,
target_resolution: str,
target_width: int,
target_height: int,
processor: dict[str, Any],
client_token: str,
) -> tuple[str, dict[str, Any]]:
payload: dict[str, Any] = {
"video_url": video_url,
"bitrate_level": processor.get("bitrate_level") or "medium",
"client_token": client_token[:64],
}
if processor_key == VideoUpscaleProcessorKey.VOLC_LARGE_MODEL_V1.value:
endpoint = "/api/v1/tools/enhance-video-generative"
normalized = str(target_resolution or "").strip().lower()
if normalized not in {"720p", "1080p", "2k"}:
raise VolcMediaKitError(
"画质增强大模型仅支持目标分辨率 720p、1080p、2K",
code="InvalidResolution",
retryable=False,
endpoint=endpoint,
)
payload["resolution"] = normalized
elif processor_key in {
VideoUpscaleProcessorKey.VOLC_STANDARD_V1.value,
VideoUpscaleProcessorKey.VOLC_PROFESSIONAL_V1.value,
}:
endpoint = "/api/v1/tools/enhance-video"
payload["tool_version"] = "standard" if processor_key == VideoUpscaleProcessorKey.VOLC_STANDARD_V1.value else "professional"
payload["resolution_limit"] = min(int(target_width), int(target_height))
if processor_key == VideoUpscaleProcessorKey.VOLC_STANDARD_V1.value:
payload["scene"] = processor.get("scene") or "aigc"
else:
raise VolcMediaKitError(
f"不支持的火山超分处理器: {processor_key}",
code="UnsupportedProcessor",
retryable=False,
)
return endpoint, payload
async def submit_video_enhance(
*,
processor_key: str,
video_url: str,
target_resolution: str,
target_width: int,
target_height: int,
processor: dict[str, Any],
client_token: str,
) -> VolcSubmitResult:
endpoint, payload = build_submit_payload(
processor_key=processor_key,
video_url=video_url,
target_resolution=target_resolution,
target_width=target_width,
target_height=target_height,
processor=processor,
client_token=client_token,
)
timeout = max(3, int(processor.get("request_timeout_seconds") or settings.VIDEO_UPSCALE_REMOTE_REQUEST_TIMEOUT_SECONDS))
try:
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
response = await client.post(f"{_base_url()}{endpoint}", headers=_headers(), json=payload)
try:
data = response.json()
except Exception:
data = {"success": False, "error": {"message": response.text[:2000]}}
if response.status_code >= 400:
raise _error_from_payload(
data,
f"火山超分提交失败 HTTP {response.status_code}",
http_status=response.status_code,
endpoint=endpoint,
)
except VolcMediaKitError:
raise
except (httpx.TimeoutException, httpx.NetworkError) as exc:
raise VolcMediaKitError(
f"火山超分提交网络异常: {exc}",
code="NetworkError",
retryable=True,
endpoint=endpoint,
) from exc
if not bool(data.get("success")) or not data.get("task_id"):
raise _error_from_payload(data, "火山超分提交失败", endpoint=endpoint)
return VolcSubmitResult(
task_id=str(data["task_id"]),
request_id=str(data.get("request_id")) if data.get("request_id") else None,
request_payload=payload,
response_payload=data,
endpoint=endpoint,
)
async def query_task(task_id: str, *, request_timeout_seconds: int | None = None) -> VolcQueryResult:
endpoint = f"/api/v1/tasks/{task_id}"
timeout = max(3, int(request_timeout_seconds or settings.VIDEO_UPSCALE_REMOTE_REQUEST_TIMEOUT_SECONDS))
try:
async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
response = await client.get(f"{_base_url()}{endpoint}", headers=_headers())
try:
data = response.json()
except Exception:
data = {"success": False, "error": {"message": response.text[:2000]}}
if response.status_code >= 400:
raise _error_from_payload(
data,
f"火山超分任务查询失败 HTTP {response.status_code}",
http_status=response.status_code,
endpoint=endpoint,
)
except VolcMediaKitError:
raise
except (httpx.TimeoutException, httpx.NetworkError) as exc:
raise VolcMediaKitError(
f"火山超分查询网络异常: {exc}",
code="NetworkError",
retryable=True,
endpoint=endpoint,
) from exc
if not bool(data.get("success")):
raise _error_from_payload(data, "火山超分任务查询失败", endpoint=endpoint)
status = str(data.get("status") or "").strip().lower()
if status not in {"running", "completed", "failed"}:
raise VolcMediaKitError(
f"火山超分返回未知任务状态: {status}",
code="UnknownStatus",
retryable=True,
request_id=str(data.get("request_id") or "") or None,
endpoint=endpoint,
response_payload=data,
)
expires_raw = data.get("expires_at")
try:
expires_at = int(expires_raw) if expires_raw is not None else None
except Exception:
expires_at = None
return VolcQueryResult(
status=status,
request_id=str(data.get("request_id")) if data.get("request_id") else None,
result=data.get("result") if isinstance(data.get("result"), dict) else None,
error=data.get("error") if isinstance(data.get("error"), dict) else None,
expires_at=expires_at,
response_payload=data,
endpoint=endpoint,
)
def is_remote_input_access_error(error: dict[str, Any] | None) -> bool:
if not error:
return False
text = " ".join(str(error.get(key) or "") for key in ("code", "message", "param", "type")).lower()
fragments = ("403", "forbidden", "url expired", "downloadfailed", "urldownloadfail", "download file", "url无法", "下载失败")
return any(fragment in text for fragment in fragments)
video-gen-api/app/tasks/celery_app.py
+40 -1
View File
@@ -18,6 +18,7 @@ CELERY_TASK_IMPORTS = (
"app.tasks.generation_poll_tasks",
"app.tasks.generation_download_tasks",
"app.tasks.generation_recovery_tasks",
"app.tasks.video_upscale_tasks",
"app.tasks.hot_opening_replicate_tasks",
"app.tasks.shot_replicate_tasks",
"app.tasks.shot_replicate_flow_tasks",
@@ -52,6 +53,14 @@ def _beat_schedule() -> dict:
"priority": settings.DOWNLOAD_TASK_PRIORITY_RECOVER,
},
}
schedule["video-upscale-recovery-every-minute"] = {
"task": CeleryTaskName.VIDEO_UPSCALE_RECOVER.value,
"schedule": 60,
"options": {
"queue": RECOVERY_QUEUE,
"priority": settings.DOWNLOAD_TASK_PRIORITY_RECOVER,
},
}
schedule["private-portrait-sync-due-assets-every-minute"] = {
"task": CeleryTaskName.PRIVATE_PORTRAIT_SYNC_DUE_ASSETS.value,
"schedule": 60,
@@ -100,10 +109,24 @@ if broker_url:
"shot_replicate.split_one_segment": {"ignore_result": True},
"shot_replicate.start_image_prompt_optimize": {"ignore_result": True},
"shot_replicate.start_video_prompt_optimize": {"ignore_result": True},
CeleryTaskName.VIDEO_UPSCALE_EXECUTE_LOCAL.value: {
"ignore_result": True,
"soft_time_limit": max(60, int(settings.VIDEO_UPSCALE_LOCAL_TIMEOUT_SECONDS or 3600)) + 60,
"time_limit": max(60, int(settings.VIDEO_UPSCALE_LOCAL_TIMEOUT_SECONDS or 3600)) + 300,
},
CeleryTaskName.VIDEO_UPSCALE_SUBMIT_REMOTE.value: {"ignore_result": True},
CeleryTaskName.VIDEO_UPSCALE_POLL_REMOTE.value: {"ignore_result": True},
CeleryTaskName.VIDEO_UPSCALE_DOWNLOAD_REMOTE_RESULT.value: {"ignore_result": True},
CeleryTaskName.VIDEO_UPSCALE_FINALIZE.value: {"ignore_result": True},
CeleryTaskName.VIDEO_UPSCALE_RECOVER.value: {"ignore_result": True},
},
worker_prefetch_multiplier=1,
broker_transport_options={
"visibility_timeout": 3600,
"visibility_timeout": max(
3600,
int(settings.VIDEO_UPSCALE_LOCAL_TIMEOUT_SECONDS or 3600) + 600,
int(settings.VIDEO_UPSCALE_REMOTE_RESULT_DOWNLOAD_TIMEOUT_SECONDS or 600) + 600,
),
"queue_order_strategy": "priority",
"priority_steps": list(range(10)),
"sep": ":",
@@ -112,6 +135,22 @@ if broker_url:
CeleryTaskName.CHATAPI_CREATE.value: {"queue": CeleryQueue.GEN_CHATAPI_CREATE.value},
CeleryTaskName.POLL_GENERATION.value: {"queue": CeleryQueue.GEN_PROVIDER_POLL.value},
CeleryTaskName.DOWNLOAD_GENERATION_RESULT.value: {"queue": CeleryQueue.GEN_RESULT_DOWNLOAD.value},
CeleryTaskName.VIDEO_UPSCALE_EXECUTE_LOCAL.value: {
"queue": settings.VIDEO_UPSCALE_LOCAL_QUEUE or CeleryQueue.GEN_VIDEO_UPSCALE_LOCAL.value
},
CeleryTaskName.VIDEO_UPSCALE_SUBMIT_REMOTE.value: {
"queue": settings.VIDEO_UPSCALE_REMOTE_QUEUE or CeleryQueue.GEN_VIDEO_UPSCALE_REMOTE.value
},
CeleryTaskName.VIDEO_UPSCALE_POLL_REMOTE.value: {
"queue": settings.VIDEO_UPSCALE_REMOTE_QUEUE or CeleryQueue.GEN_VIDEO_UPSCALE_REMOTE.value
},
CeleryTaskName.VIDEO_UPSCALE_DOWNLOAD_REMOTE_RESULT.value: {
"queue": settings.VIDEO_UPSCALE_REMOTE_QUEUE or CeleryQueue.GEN_VIDEO_UPSCALE_REMOTE.value
},
CeleryTaskName.VIDEO_UPSCALE_FINALIZE.value: {
"queue": settings.VIDEO_UPSCALE_LOCAL_QUEUE or CeleryQueue.GEN_VIDEO_UPSCALE_LOCAL.value
},
CeleryTaskName.VIDEO_UPSCALE_RECOVER.value: {"queue": RECOVERY_QUEUE},
CeleryTaskName.DISPATCH_DUE_POLL.value: {"queue": RECOVERY_QUEUE},
"hot_opening.start_image_prompt_optimize": {"queue": CeleryQueue.GEN_CHATAPI_CREATE.value},
"hot_opening.start_video_prompt_optimize": {"queue": CeleryQueue.GEN_CHATAPI_CREATE.value},
video-gen-api/app/tasks/generation_create_tasks.py
+1 -1
View File
@@ -91,7 +91,7 @@ def _build_optimized_prompt_by_params(task: ChatGenerationTask) -> str:
duration = _get_first_value(task, "duration")
aspect_ratio = _get_first_value(task, "aspect_ratio")
resolution = _get_first_value(task, "resolution")
resolution = _get_first_value(task, "provider_generation_resolution", "resolution")
image_size = _get_first_value(task, "image_size")
image_px = _get_first_value(task, "image_px")
image_proportion = _get_first_value(task, "image_proportion")
video-gen-api/app/tasks/generation_download_tasks.py
+42 -2
View File
@@ -29,7 +29,7 @@ from app.services.celery_download_recovery_service import (
upsert_download_active,
)
from app.services.error_codes import extract_error_message
from app.services.generation.download_service import download_generation_result
from app.services.generation.download_service import download_generation_result, download_video_upscale_source
from app.services.generation.log_service import log_task_event
from app.services.generation.refund_service import mark_chat_generation_task_failed_and_refund_once
from app.services.media_token_usage_snapshot_service import sync_chat_generation_task_media_token_snapshot
@@ -560,7 +560,16 @@ async def _run(task_id: str):
return
try:
downloaded = await download_generation_result(task)
use_video_upscale = bool(
task.gen_type == GenerationType.VIDEO.value
and task.video_upscale_enabled_snapshot
and task.video_upscale_snapshot_json
)
downloaded = (
await download_video_upscale_source(task)
if use_video_upscale
else await download_generation_result(task)
)
task = await _reload_task(db, task_id)
if not task:
@@ -572,6 +581,37 @@ async def _run(task_id: str):
await remove_download_active(task_id)
return
if use_video_upscale:
from app.services.video_upscale.task_service import prepare_video_upscale_task, enqueue_upscale_task
upscale = await prepare_video_upscale_task(
db,
task=task,
source_local_path=str(downloaded.storage_path or ""),
source_file_size_bytes=downloaded.file_size_bytes,
)
task.download_lease_until = None
task.download_next_retry_at = None
task.download_last_error = None
task.retry_count = 0
await db.commit()
await remove_download_active(task.id)
await enqueue_upscale_task(db, upscale=upscale, reason="source_download_completed")
await _log_download_event(
task,
event_type=ChatGenerationTaskEventType.DOWNLOAD_SUCCESS,
to_status=ChatGenerationTaskStatus.GENERATING.value,
to_stage=ChatGenerationPipelineStage.UPSCALE_QUEUED.value,
detail={
"upscale_source_path": downloaded.storage_path,
"file_size_bytes": downloaded.file_size_bytes,
"download_attempt_count": task.download_attempt_count,
"user_resource_recorded": False,
"cover_generated": False,
},
)
return
if task.gen_type == GenerationType.IMAGE.value:
task.image_url = downloaded.url
else:
video-gen-api/app/tasks/generation_recovery_tasks.py
+15
View File
@@ -52,6 +52,13 @@ async def _run_shot_split_once() -> Dict[str, Any]:
return await recover_shot_split_tasks_once(db)
async def _run_video_upscale_once() -> Dict[str, Any]:
from app.services.video_upscale.task_service import recover_video_upscale_tasks_once
async with async_session() as db:
return await recover_video_upscale_tasks_once(db)
async def _run_with_execution_lock(
*,
lock_key: str,
@@ -155,6 +162,8 @@ async def _run_startup_recovery_once() -> Dict[str, Any]:
- 创建/提词/视频分析 -> gen_chatapi_create
- provider poll -> gen_provider_poll
- 下载/ffmpeg 切片 -> gen_result_download
- 本地视频超分 -> gen_video_upscale_local
- 火山视频超分 -> gen_video_upscale_remote
恢复扫描本身只走 gen_recovery，避免堵住业务 worker。
"""
return await _run_with_execution_lock(
@@ -192,6 +201,12 @@ async def _run_startup_recovery_steps() -> Dict[str, Any]:
"download_recovery",
_run_download_once,
),
(
"video_upscale",
settings.VIDEO_UPSCALE_RECOVERY_LOCK_KEY,
"video_upscale_recovery",
_run_video_upscale_once,
),
]
for name, lock_key, log_context, runner in steps:
video-gen-api/app/tasks/video_upscale_tasks.py
+109
View File
@@ -0,0 +1,109 @@
from __future__ import annotations
from typing import Any
from app.config import settings
from app.models.base import async_session
from app.services.video_upscale.task_service import (
recover_video_upscale_tasks_once,
run_finalize_upscale,
run_local_upscale,
run_remote_poll,
run_remote_result_download,
run_remote_submit,
)
from app.tasks.async_runner import run_async
from app.tasks.celery_app import celery_app
async def _run_local(upscale_task_id: str) -> None:
async with async_session() as db:
await run_local_upscale(db, upscale_task_id)
async def _run_submit(upscale_task_id: str, *, count_attempt: bool = True) -> None:
async with async_session() as db:
await run_remote_submit(db, upscale_task_id, count_attempt=count_attempt)
async def _run_poll(upscale_task_id: str) -> None:
async with async_session() as db:
await run_remote_poll(db, upscale_task_id)
async def _run_download(upscale_task_id: str) -> None:
async with async_session() as db:
await run_remote_result_download(db, upscale_task_id)
async def _run_finalize(upscale_task_id: str) -> None:
async with async_session() as db:
await run_finalize_upscale(db, upscale_task_id)
async def _run_recovery() -> dict[str, Any]:
async with async_session() as db:
return await recover_video_upscale_tasks_once(db)
if celery_app:
@celery_app.task(name="video_upscale.execute_local", bind=True, max_retries=2)
def execute_local(self, upscale_task_id: str) -> None:
try:
return run_async(_run_local(upscale_task_id))
except Exception as exc:
raise self.retry(exc=exc, countdown=max(5, int(settings.VIDEO_UPSCALE_RETRY_BACKOFF_SECONDS or 60)))
@celery_app.task(name="video_upscale.submit_remote", bind=True, max_retries=2)
def submit_remote(self, upscale_task_id: str, count_attempt: bool = True) -> None:
try:
return run_async(_run_submit(upscale_task_id, count_attempt=count_attempt))
except Exception as exc:
raise self.retry(exc=exc, countdown=max(5, int(settings.VIDEO_UPSCALE_RETRY_BACKOFF_SECONDS or 60)))
@celery_app.task(name="video_upscale.poll_remote", bind=True, max_retries=2)
def poll_remote(self, upscale_task_id: str) -> None:
try:
return run_async(_run_poll(upscale_task_id))
except Exception as exc:
raise self.retry(exc=exc, countdown=max(5, int(settings.VIDEO_UPSCALE_RETRY_BACKOFF_SECONDS or 60)))
@celery_app.task(name="video_upscale.download_remote_result", bind=True, max_retries=2)
def download_remote_result(self, upscale_task_id: str) -> None:
try:
return run_async(_run_download(upscale_task_id))
except Exception as exc:
raise self.retry(exc=exc, countdown=max(5, int(settings.VIDEO_UPSCALE_RETRY_BACKOFF_SECONDS or 60)))
@celery_app.task(name="video_upscale.finalize", bind=True, max_retries=2)
def finalize(self, upscale_task_id: str) -> None:
try:
return run_async(_run_finalize(upscale_task_id))
except Exception as exc:
raise self.retry(exc=exc, countdown=max(5, int(settings.VIDEO_UPSCALE_RETRY_BACKOFF_SECONDS or 60)))
@celery_app.task(name="video_upscale.recover_once", bind=True)
def recover_once(self) -> dict[str, Any]:
return run_async(_run_recovery())
else:
class _DisabledTask:
def delay(self, *args: Any, **kwargs: Any) -> None:
raise RuntimeError("Celery is disabled")
def apply_async(self, *args: Any, **kwargs: Any) -> None:
raise RuntimeError("Celery is disabled")
execute_local = _DisabledTask()
submit_remote = _DisabledTask()
poll_remote = _DisabledTask()
download_remote_result = _DisabledTask()
finalize = _DisabledTask()
recover_once = _DisabledTask()
video-gen-api/app/types/generation/provider.py
+1
View File
@@ -14,6 +14,7 @@ class ProviderGenerationRecordLike(Protocol):
duration: int | None
aspect_ratio: str | None
resolution: str | None
provider_generation_resolution: str | None
image_size: str | None
image_proportion: str | None
image_px: str | None