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

冲突解决

Browse Source
This commit is contained in:
sjy
2026-07-02 14:54:57 +08:00
parent b96e4ec4bf 6b7604dd38
commit 86fff5fd83
48 changed files with 4275 additions and 2023 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-admin/dist/assets/index-DfKYC1kL.js
Vendored
+548
View File
File diff suppressed because one or more lines are too long
video-gen-admin/dist/index.html
Vendored
+1 -1
View File
@@ -28,7 +28,7 @@
}
})();
</script>
<script type="module" crossorigin src="/assets/index-DveJ8Oia.js"></script>
<script type="module" crossorigin src="/assets/index-DfKYC1kL.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-D7ShJUt4.css">
</head>
<body>
video-gen-admin/src/pages/AdminVideoEngines.tsx
+27 -5
View File
@@ -20,6 +20,8 @@ interface VideoEngine {
maxDuration: number;
maxImageCount: number;
maxVideoCount: number;
supportsFirstLastFrame: boolean;
supportsUniversalReference: boolean;
isActive: boolean;
priority: number;
}
@@ -72,6 +74,8 @@ const AdminVideoEngines: React.FC = () => {
max_duration: values.maxDuration ?? 15,
max_image_count: values.maxImageCount ?? 2,
max_video_count: values.maxVideoCount ?? 0,
supports_first_last_frame: values.supportsFirstLastFrame ?? false,
supports_universal_reference: values.supportsUniversalReference ?? true,
is_active: values.isActive ?? true,
priority: values.priority ?? 0,
};
@@ -112,6 +116,8 @@ const AdminVideoEngines: React.FC = () => {
maxDuration: 15,
maxImageCount: 2,
maxVideoCount: 0,
supportsFirstLastFrame: false,
supportsUniversalReference: true,
supportedRatios: ['16:9', '4:3', '1:1', '3:4', '9:16', '21:9'],
supportedResolutions: ['480p', '720p', '1080p'],
supportedDurations: [4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],
@@ -159,6 +165,14 @@ const AdminVideoEngines: React.FC = () => {
title: '最大视频', dataIndex: 'maxVideoCount', width: 100,
render: (v: number) => <Tag color="cyan">{v} 个</Tag>,
},
{
title: '首尾帧', dataIndex: 'supportsFirstLastFrame', width: 90,
render: (v: boolean) => <Tag color={v ? 'green' : 'default'}>{v ? '支持' : '不支持'}</Tag>,
},
{
title: '全能参考', dataIndex: 'supportsUniversalReference', width: 100,
render: (v: boolean) => <Tag color={v ? 'purple' : 'default'}>{v ? '支持' : '不支持'}</Tag>,
},
{
title: '状态', dataIndex: 'isActive', width: 80,
render: (v: boolean) => <Tag color={v ? 'green' : 'default'}>{v ? '启用' : '停用'}</Tag>,
@@ -196,7 +210,7 @@ const AdminVideoEngines: React.FC = () => {
rowKey="id"
loading={loading}
pagination={false}
scroll={{ x: 900 }}
scroll={{ x: 1100 }}
/>
</Card>
@@ -263,7 +277,18 @@ const AdminVideoEngines: React.FC = () => {
</Form.Item>
</div>
<div style={{ display: 'flex', gap: 16 }}>
<Form.Item name="priority" label="优先级">
<Form.Item name="supportsFirstLastFrame" label="首尾帧模式" valuePropName="checked" style={{ paddingTop: 30, flex: 1 }}>
<Switch checkedChildren="支持" unCheckedChildren="不支持" />
</Form.Item>
<Form.Item name="supportsUniversalReference" label="全能参考模式" valuePropName="checked" style={{ paddingTop: 30, flex: 1 }}>
<Switch checkedChildren="支持" unCheckedChildren="不支持" />
</Form.Item>
<Form.Item name="isActive" label="启用状态" valuePropName="checked" style={{ paddingTop: 30, flex: 1 }}>
<Switch />
</Form.Item>
</div>
<div style={{ display: 'flex', gap: 16 }}>
<Form.Item name="priority" label="优先级" style={{ flex: 1 }}>
<Select size="large" options={[
{ value: 0, label: '0 (默认)' },
{ value: 1, label: '1' },
@@ -273,9 +298,6 @@ const AdminVideoEngines: React.FC = () => {
{ value: 10, label: '10 (最高)' },
]} />
</Form.Item>
<Form.Item name="isActive" label="启用状态" valuePropName="checked" style={{ paddingTop: 30 }}>
<Switch />
</Form.Item>
</div>
</Form>
</Modal>
video-gen-api/alembic/versions/d4e5f6a7b8c9_add_first_last_frame_universal_reference.py
+46
View File
@@ -0,0 +1,46 @@
"""d4e5f6a7b8c9 - 视频引擎增加首尾帧和全能参考支持字段
Revision ID: d4e5f6a7b8c9
Revises: 6idufv2q1c
Create Date: 2026-07-02 00:00:00.000000
该文件包含 2026-07-02 的数据库迁移内容：
1. 视频引擎表增加 supports_first_last_frame 字段（是否支持首尾帧模式）
2. 视频引擎表增加 supports_universal_reference 字段（是否支持全能参考模式）
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = 'd4e5f6a7b8c9'
down_revision: Union[str, None] = 'e2618c38eae5'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.add_column(
'video_engines',
sa.Column(
'supports_first_last_frame',
sa.Boolean(),
server_default=sa.text('false'),
nullable=False,
),
)
op.add_column(
'video_engines',
sa.Column(
'supports_universal_reference',
sa.Boolean(),
server_default=sa.text('true'),
nullable=False,
),
)
def downgrade() -> None:
op.drop_column('video_engines', 'supports_universal_reference')
op.drop_column('video_engines', 'supports_first_last_frame')
video-gen-api/alembic/versions/e2618c38eae5_merge_changes_main.py
+25
View File
@@ -0,0 +1,25 @@
"""merge changes main
Revision ID: e2618c38eae5
Revises: 6idufv2q1c, ee4b42e57960
Create Date: 2026-07-02 09:43:29.076528
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'e2618c38eae5'
down_revision: Union[str, None] = ('6idufv2q1c', 'ee4b42e57960')
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
video-gen-api/alembic/versions/ee4b42e57960_add_chat_generation_poll_schedule_fields.py
+83
View File
@@ -0,0 +1,83 @@
"""add chat generation poll schedule fields
Revision ID: ee4b42e57960
Revises: f7a3b2c1d4e5
Create Date: 2026-07-01 16:11:28.799227
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'ee4b42e57960'
down_revision: Union[str, None] = 'f7a3b2c1d4e5'
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('poll_started_at', sa.DateTime(timezone=True), nullable=True),
)
op.add_column(
'chat_generation_tasks',
sa.Column('next_poll_at', sa.DateTime(timezone=True), nullable=True),
)
op.add_column(
'chat_generation_tasks',
sa.Column(
'poll_interval_seconds',
sa.Integer(),
nullable=False,
server_default='0',
comment='当前轮询退避间隔秒数',
),
)
# 修复历史“生成中视频任务”：
# 1. deadline_at 改为 created_at + 24 小时，避免沿用旧 30 分钟最终超时。
# 2. next_poll_at 设置为 now()，让 Beat dispatcher 能尽快接管。
# 3. poll_started_at 用 last_poll_at / created_at 兜底。
op.execute(
"""
UPDATE chat_generation_tasks
SET
poll_started_at = COALESCE(poll_started_at, last_poll_at, created_at, NOW()),
next_poll_at = COALESCE(next_poll_at, NOW()),
poll_interval_seconds = COALESCE(poll_interval_seconds, 0),
deadline_at = CASE
WHEN deadline_at IS NULL THEN created_at + INTERVAL '24 hours'
WHEN deadline_at < created_at + INTERVAL '24 hours' THEN created_at + INTERVAL '24 hours'
ELSE deadline_at
END
WHERE deleted_at IS NULL
AND status = 'generating'
AND gen_type = 'video'
"""
)
op.create_index(
'idx_chat_generation_tasks_next_poll_at',
'chat_generation_tasks',
['next_poll_at'],
unique=False,
postgresql_where=sa.text(
"deleted_at IS NULL "
"AND status = 'generating' "
"AND gen_type = 'video' "
"AND next_poll_at IS NOT NULL"
),
)
def downgrade() -> None:
op.drop_index(
'idx_chat_generation_tasks_next_poll_at',
table_name='chat_generation_tasks',
)
op.drop_column('chat_generation_tasks', 'poll_interval_seconds')
op.drop_column('chat_generation_tasks', 'next_poll_at')
op.drop_column('chat_generation_tasks', 'poll_started_at')
video-gen-api/app/api/v1/admin.py
+1 -1
View File
@@ -995,7 +995,7 @@ async def list_credit_ratios(
admin: User = Depends(get_admin_user),
db: AsyncSession = Depends(get_db),
):
result = await db.execute(select(CreditRatio))
result = await db.execute(select(CreditRatio).order_by(CreditRatio.gen_type.desc(), CreditRatio.model_config_id.desc()))
return result.scalars().all()
video-gen-api/app/api/v1/generation_ai.py
+52
View File
@@ -9,6 +9,8 @@ from app.models.chat_generation_task import ChatGenerationTask
from app.models.user import User
from app.schemas.generation_ai import (
GenerationAIEngineOptionsOut,
GenerationAIHistoryBatchDeleteOut,
GenerationAIHistoryBatchDeleteRequest,
GenerationAIHistoryDayItemsOut,
GenerationAIHistoryGroupedOut,
GenerationAIRetryOut,
@@ -31,6 +33,7 @@ from app.services.generation_billing_service import (
charge_generation_media_by_params,
get_next_credit_attempt_no,
)
from app.services.generation_history_delete_service import batch_delete_generation_history_items
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.resource_capacity_service import assert_user_resource_capacity_available
@@ -87,6 +90,8 @@ router = APIRouter(
"supported_durations": [4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15],
"max_duration": 15,
"priority": 10,
"supports_first_last_frame": False,
"supports_universal_reference": False,
}
],
}
@@ -332,6 +337,53 @@ async def list_history_grouped_days(
)
@router.delete(
"/history/batch",
response_model=GenerationAIHistoryBatchDeleteOut,
summary="批量删除素材云历史记录",
description=(
"按 history_source 批量软删除素材云历史记录，单次最多30条。"
"generation_record 和 chat_task 入参 ids 为对应记录ID，且只有生成完成后才能删除；"
"hot_opening_replicate 入参 ids 为 module_project_id；"
"shot_replicate 入参 ids 为 shot_segment_id。"
"爆款开头复刻和拆镜复刻会联动软删 ModuleGenerationProject、ModuleGenerationStep、关联 ChatGenerationTask 和 generated_resources；"
"拆镜复刻还会联动软删 ShotReplicateSegment。"
"如果存在生成中、轮询中、下载中等任务，接口直接拦截，不做失败标记、不退款。"
),
responses={
200: {
"description": "批量软删除成功",
},
400: {
"description": "参数错误，例如 history_source 不支持、ids 为空、超过30条或重复",
},
401: {
"description": "未登录或 Token 无效",
},
404: {
"description": "部分 ID 不存在、不属于当前用户或已删除",
},
409: {
"description": "存在未完成或生成中的记录，当前不能删除",
},
},
)
async def batch_delete_history_items(
req: GenerationAIHistoryBatchDeleteRequest = Body(
...,
description="素材云历史批量删除参数。不同 history_source 对应不同 ID 语义，详见字段说明",
),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
return await batch_delete_generation_history_items(
db=db,
current_user=current_user,
history_source=req.history_source,
ids=req.ids,
)
@router.get(
"/history/{generated_date}",
response_model=GenerationAIHistoryDayItemsOut,
video-gen-api/app/api/v1/shot_replicate.py
+30
View File
@@ -31,6 +31,7 @@ from app.schemas.shot_replicate import (
ShotReplicateSpecOut,
ShotReplicateTaskDetailOut,
ShotReplicateVideoPromptSchemaUpdateRequest,
ShotSegmentDeleteOut,
ShotSegmentDetailOut,
ShotSegmentListOut,
ShotSegmentReplicationCreateRequest,
@@ -60,6 +61,7 @@ from app.services.shot_replicate_taskset_service import (
create_custom_segment,
create_segments_by_ai,
create_task_set,
delete_segment,
get_segment_for_user,
list_segments,
list_task_sets,
@@ -455,6 +457,34 @@ async def get_segment(
return await segment_detail(db, current_user=current_user, segment_id=segment_id)
@router.delete(
"/segments/{segment_id}",
response_model=ShotSegmentDeleteOut,
summary="软删除拆镜片段",
description=(
"软删除单个拆镜片段；不删除 segment_video_path 指向的物理文件。"
"如片段已创建拆镜复刻项目，会联动软删除该项目和已生成资源账本，但用户主动删除不退款。"
"如片段或关联项目仍有处理中任务，会拒绝删除。"
),
)
async def delete_shot_segment(
segment_id: str = Path(..., description="拆镜片段ID，即 shot_replicate_segments.id"),
current_user: User = Depends(get_current_user),
db: AsyncSession = Depends(get_db),
):
try:
out = await delete_segment(db, current_user=current_user, segment_id=segment_id)
await db.commit()
return out
except HTTPException:
await db.rollback()
raise
except Exception as exc:
await db.rollback()
_log_api_exception_from_locals(exc, locals(), f"删除拆镜片段失败: {exc}")
raise HTTPException(status_code=500, detail=f"删除拆镜片段失败: {exc}")
@router.post(
"/segments/{segment_id}/replication-projects",
response_model=ShotReplicateActionOut,
video-gen-api/app/api/v1/video_engines.py
+2
View File
@@ -51,5 +51,7 @@ async def list_active_engines(
"supported_durations": durations,
"max_image_count": e.max_image_count,
"max_video_count": e.max_video_count,
"supports_first_last_frame": e.supports_first_last_frame,
"supports_universal_reference": e.supports_universal_reference,
})
return {"items": items}
video-gen-api/app/config.py
+16 -1
View File
@@ -121,7 +121,14 @@ class Settings(BaseSettings):
CHATAPI_ASYNC_RETRY_BACKOFF_SECONDS: int = 30
CHATAPI_ASYNC_POLL_INTERVAL_SECONDS: int = 30
CHATAPI_ASYNC_IMAGE_DEADLINE_MINUTES: int = 10
CHATAPI_ASYNC_VIDEO_DEADLINE_MINUTES: int = 30
# 视频异步生成不再使用 30 分钟最终超时；前 10 分钟高频轮询，之后降频，24 小时最后判定失败才退款。
CHATAPI_ASYNC_VIDEO_FINAL_DEADLINE_HOURS: int = 24
CHATAPI_ASYNC_VIDEO_HIGH_FREQ_MINUTES: int = 10
CHATAPI_ASYNC_VIDEO_HIGH_FREQ_POLL_SECONDS: int = 30
CHATAPI_ASYNC_VIDEO_BACKOFF_INITIAL_SECONDS: int = 60
CHATAPI_ASYNC_VIDEO_BACKOFF_MULTIPLIER: int = 2
CHATAPI_ASYNC_VIDEO_BACKOFF_MAX_SECONDS: int = 3600
CHATAPI_ASYNC_VIDEO_DIRECT_COUNTDOWN_MAX_SECONDS: int = 300
# Distributed provider concurrency limits. 0 means disabled/no-op.
ARK_CHAT_PROMPT_MAX_CONCURRENCY: int = 20
@@ -184,6 +191,14 @@ class Settings(BaseSettings):
MODULE_ASYNC_RECOVERY_LOCK_KEY: str = "vg:celery:module_async_recovery_lock"
SHOT_SPLIT_RECOVERY_LOCK_KEY: str = "vg:celery:shot_split_recovery_lock"
# 视频到期轮询调度。
# Celery Beat 每分钟投递轻量 dispatcher 到 gen_recovery；dispatcher 只扫描 next_poll_at 到期的视频任务。
POLL_DUE_DISPATCH_ENABLED: bool = True
POLL_DUE_DISPATCH_INTERVAL_SECONDS: int = 60
POLL_DUE_DISPATCH_BATCH_SIZE: int = 100
POLL_DUE_DISPATCH_LOCK_KEY: str = "vg:celery:poll_due_dispatch_lock"
POLL_DUE_DISPATCH_LOCK_TTL_SECONDS: int = 55
# 模块异步任务容灾配置。
# 覆盖 ModuleGenerationStep 提词任务、shot 原视频/片段分析、shot ffmpeg 切割 active 注册。
# 恢复扫描走 CELERY_RECOVERY_QUEUE，真实业务任务回到原始队列。
video-gen-api/app/enums/__init__.py
+1
View File
@@ -14,3 +14,4 @@ from app.enums.notification import *
from app.enums.resource_capacity import *
from app.enums.team import *
from app.enums.home_material import *
from app.enums.celery_queue import *
video-gen-api/app/enums/celery_queue.py
+21
View File
@@ -0,0 +1,21 @@
from enum import Enum
class CeleryQueue(str, Enum):
GEN_CHATAPI_CREATE = "gen_chatapi_create"
GEN_PROVIDER_POLL = "gen_provider_poll"
GEN_RESULT_DOWNLOAD = "gen_result_download"
GEN_RECOVERY = "gen_recovery"
DEFAULT = "default"
class CeleryTaskName(str, Enum):
CHATAPI_CREATE = "generation.chatapi_create_generation_task"
POLL_GENERATION = "generation.poll_generation_task"
DOWNLOAD_GENERATION_RESULT = "generation.download_generation_result_task"
RECOVER_DOWNLOAD = "generation.recover_download_tasks_once"
RECOVER_GENERATION = "generation.recover_generation_tasks_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"
SHOT_SPLIT_RECOVERY = "shot_replicate.recover_split_tasks_once"
video-gen-api/app/enums/generation_history.py
+3
View File
@@ -118,3 +118,6 @@ def is_generation_history_module_source(source: GenerationHistorySourceEnum) ->
"""判断当前来源是否需要回填模块项目信息。"""
return source in GENERATION_HISTORY_MODULE_SOURCES
MAX_BATCH_DELETE_COUNT = 30
video-gen-api/app/enums/generation_task.py
+10
View File
@@ -45,12 +45,18 @@ class ChatGenerationTaskEventType(str, Enum):
POLL_START = "POLL_START"
POLL_PENDING = "POLL_PENDING"
POLL_SCHEDULED = "POLL_SCHEDULED"
POLL_SKIP_NOT_DUE = "POLL_SKIP_NOT_DUE"
POLL_DISPATCH_DUE = "POLL_DISPATCH_DUE"
POLL_DISPATCH_SKIP = "POLL_DISPATCH_SKIP"
POLL_SUCCESS = "POLL_SUCCESS"
POLL_FAILED = "POLL_FAILED"
POLL_SUCCESS_AFTER_TIMEOUT_RECOVERY = "POLL_SUCCESS_AFTER_TIMEOUT_RECOVERY"
FINAL_POLL_BEFORE_TIMEOUT = "FINAL_POLL_BEFORE_TIMEOUT"
FINAL_POLL_BEFORE_TIMEOUT_ERROR = "FINAL_POLL_BEFORE_TIMEOUT_ERROR"
FINAL_POLL_BEFORE_TIMEOUT_PENDING = "FINAL_POLL_BEFORE_TIMEOUT_PENDING"
GENERATION_RECOVERY_ENQUEUE = "GENERATION_RECOVERY_ENQUEUE"
GENERATION_RECOVERY_TIMEOUT = "GENERATION_RECOVERY_TIMEOUT"
DOWNLOAD_ENQUEUE = "DOWNLOAD_ENQUEUE"
DOWNLOAD_ENQUEUE_FAILED = "DOWNLOAD_ENQUEUE_FAILED"
@@ -78,6 +84,7 @@ class ChatGenerationTaskEventType(str, Enum):
DOWNLOAD_SKIP_DISABLED = "DOWNLOAD_SKIP_DISABLED"
TASK_TIMEOUT = "TASK_TIMEOUT"
TASK_FAILED = "TASK_FAILED"
ALLOWED_GENERATION_MODES = {
@@ -99,3 +106,6 @@ DOWNLOAD_RECOVERABLE_STAGES = {
ChatGenerationPipelineStage.DOWNLOADING.value,
ChatGenerationPipelineStage.RETRY_WAITING.value,
}
PROVIDER_SUCCESS_STATUSES = {"succeeded", "success", "completed", "done"}
PROVIDER_FAILED_STATUSES = {"failed", "error", "canceled", "cancelled"}
video-gen-api/app/models/__init__.py
+2 -1
View File
@@ -31,6 +31,7 @@ from app.models.user_oauth import UserOAuth
from app.models.user_oauth_account import UserOAuthAccount
from app.models.user_oauth_app import UserOAuthApp
from app.models.home_material import HomeMaterialAsset, HomeMaterialCategory, HomeMaterialWatermark
from app.models.contact_request import ContactRequest
__all__ = [
"Base", "TimestampMixin", "SoftDeleteMixin", "engine", "async_session",
@@ -38,7 +39,7 @@ __all__ = [
"User", "Team", "Project", "GenerationRecord", "CreditRecord",
"ModelConfig", "SystemConfig", "Notification", "PaymentOrder",
"TokenUsage", "IndustryConfig", "VideoEngine", "CreditRatio",
"MenuConfig", "RechargePackage", "OperationLog",
"MenuConfig", "RechargePackage", "OperationLog", "ContactRequest",
"ChatGenerationTask", "ChatGenerationTaskEvent", "ChatProviderCallLog",
"GeneratedResource", "UserResourceMonthStat", "UserResourceTotalStat",
"UserResourceCapacityConfig",
video-gen-api/app/models/chat_generation_task.py
+16
View File
@@ -26,6 +26,17 @@ class ChatGenerationTask(Base, TimestampMixin, SoftDeleteMixin):
unique=True,
postgresql_where=text("deleted_at IS NULL AND idempotency_key IS NOT NULL"),
),
# 视频 24 小时降频轮询调度使用。
Index(
"idx_chat_generation_tasks_next_poll_at",
"next_poll_at",
postgresql_where=text(
"deleted_at IS NULL "
"AND status = 'generating' "
"AND gen_type = 'video' "
"AND next_poll_at IS NOT NULL"
),
),
)
@@ -72,6 +83,11 @@ class ChatGenerationTask(Base, TimestampMixin, SoftDeleteMixin):
retry_count: Mapped[int] = mapped_column(Integer, default=0)
poll_count: Mapped[int] = mapped_column(Integer, default=0)
last_poll_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
# 视频降频轮询调度字段。
# 图片同步生成仍沿用原超时逻辑；这些字段主要给 video + provider poll 使用。
poll_started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
next_poll_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
poll_interval_seconds: Mapped[int] = mapped_column(Integer, default=0)
deadline_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
generated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
video-gen-api/app/models/video_engine.py
+2
View File
@@ -19,6 +19,8 @@ class VideoEngine(Base, TimestampMixin):
max_duration: Mapped[int] = mapped_column(Integer, default=15)
max_image_count: Mapped[int] = mapped_column(Integer, default=2)
max_video_count: Mapped[int] = mapped_column(Integer, default=0)
supports_first_last_frame: Mapped[bool] = mapped_column(Boolean, default=False)
supports_universal_reference: Mapped[bool] = mapped_column(Boolean, default=True)
generate_url: Mapped[str | None] = mapped_column(String(512), nullable=True, default="")
query_url: Mapped[str | None] = mapped_column(String(512), nullable=True, default="")
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
video-gen-api/app/schemas/generation_ai.py
+95 -1
View File
@@ -1,8 +1,9 @@
from typing import Annotated, Literal
from pydantic import BaseModel, ConfigDict, Field
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.schemas.common import NaiveDatetimeOptional
from app.enums.generation_history import normalize_generation_history_source
class GenerationAIReference(BaseModel):
@@ -192,6 +193,8 @@ class GenerationAIVideoEngineOptionOut(BaseModel):
priority: int = Field(0, description="引擎优先级，数值越大越优先")
max_image_count: int | None = Field(None, description="最大图片数量")
max_video_count: int | None = Field(None, description="最大视频数量")
supports_first_last_frame: bool = Field(False, description="是否支持首帧和最后一帧")
supports_universal_reference: bool = Field(False, description="是否支持通用参考")
class GenerationAIEngineGroupOut(BaseModel):
@@ -245,6 +248,8 @@ class GenerationAIEngineOptionsOut(BaseModel):
"priority": 10,
"max_image_count": 2,
"max_video_count": 0,
"supports_first_last_frame": False,
"supports_universal_reference": False,
}
],
}
@@ -492,6 +497,95 @@ class GenerationAITaskDeleteOut(BaseModel):
freed_size_bytes: int = Field(0, description="本次软删联动释放的有效资源空间字节数")
class GenerationAIHistoryBatchDeleteRequest(BaseModel):
"""素材云历史批量删除请求体。"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"history_source": "shot_replicate",
"ids": ["shot_segment_id_1", "shot_segment_id_2"],
}
}
)
history_source: str = Field(
...,
description=(
"素材云历史来源。"
"generation_record=项目生成，chat_task=AI创作，"
"hot_opening_replicate=爆款开头复刻，shot_replicate=拆镜复刻。"
"注意：项目生成/AI创作传记录ID；爆款开头复刻传 module_project_id；拆镜复刻传 shot_segment_id。"
),
examples=["shot_replicate"],
)
ids: list[str] = Field(
...,
min_length=1,
max_length=30,
description="需要删除的ID数组，最多30个；不允许重复或空字符串",
examples=[["shot_segment_id_1", "shot_segment_id_2"]],
)
@field_validator("history_source")
@classmethod
def validate_history_source(cls, value: str) -> str:
try:
return normalize_generation_history_source(value).value
except ValueError as exc:
raise ValueError("history_source 不支持") from exc
@field_validator("ids")
@classmethod
def validate_ids(cls, values: list[str]) -> list[str]:
normalized = [str(item).strip() for item in values if str(item or "").strip()]
if not normalized:
raise ValueError("ids 不能为空")
if len(normalized) > 30:
raise ValueError("单次最多删除30条记录")
if len(normalized) != len(set(normalized)):
raise ValueError("ids 不允许重复")
return normalized
class GenerationAIHistoryBatchDeleteOut(BaseModel):
"""素材云历史批量删除响应体。"""
model_config = ConfigDict(
json_schema_extra={
"example": {
"message": "删除成功",
"history_source": "shot_replicate",
"history_source_label": "拆镜复刻",
"requested_count": 2,
"deleted_count": 2,
"requested_ids": ["shot_segment_id_1", "shot_segment_id_2"],
"deleted_ids": ["shot_segment_id_1", "shot_segment_id_2"],
"generation_record_ids": [],
"chat_task_ids": ["chat_task_id_1", "chat_task_id_2"],
"module_project_ids": ["module_project_id_1", "module_project_id_2"],
"shot_segment_ids": ["shot_segment_id_1", "shot_segment_id_2"],
"deleted": True,
"freed_size_bytes": 123456,
}
}
)
message: str = Field(..., description="操作结果提示信息")
history_source: str = Field(..., description="素材云历史来源")
history_source_label: str | None = Field(None, description="素材云历史来源中文名称")
requested_count: int = Field(..., description="请求删除数量")
deleted_count: int = Field(..., description="实际删除数量")
requested_ids: list[str] = Field(default_factory=list, description="请求删除的原始ID列表")
deleted_ids: list[str] = Field(default_factory=list, description="已删除的原始ID列表")
generation_record_ids: list[str] = Field(default_factory=list, description="联动软删除的 GenerationRecord ID")
chat_task_ids: list[str] = Field(default_factory=list, description="联动软删除的 ChatGenerationTask ID")
module_project_ids: list[str] = Field(default_factory=list, description="联动软删除的 ModuleGenerationProject ID")
shot_segment_ids: list[str] = Field(default_factory=list, description="联动软删除的 ShotReplicateSegment ID")
deleted: bool = Field(..., description="是否已完成软删除")
freed_size_bytes: int = Field(0, description="本次软删联动释放的有效资源空间字节数")
class GenerationAIRetryOut(BaseModel):
"""AI生成任务重试响应体。"""
video-gen-api/app/schemas/shot_replicate.py
+10
View File
@@ -484,6 +484,16 @@ class ShotReplicateDeleteOut(BaseModel):
message: str = Field(..., description="删除结果提示")
project_id: str = Field(..., description="被软删除的总任务项目ID")
deleted: bool = Field(..., description="是否已软删除")
released_size_bytes: int = Field(0, description="本次软删释放的用户容量占用字节数；不删除物理文件")
class ShotSegmentDeleteOut(BaseModel):
message: str = Field(..., description="删除结果提示")
segment_id: str = Field(..., description="被软删除的拆镜片段ID")
task_set_id: str = Field(..., description="所属拆镜总任务集ID")
deleted: bool = Field(..., description="是否已软删除")
deleted_module_project_id: str | None = Field(None, description="联动软删除的拆镜复刻项目ID；没有关联项目时为空")
released_size_bytes: int = Field(0, description="本次释放的用户容量占用字节数；只释放数据账本，不删除物理文件")
video-gen-api/app/schemas/video_engine.py
+4
View File
@@ -15,6 +15,8 @@ class VideoEngineCreate(BaseModel):
max_duration: int = Field(default=15)
max_image_count: int = Field(default=2)
max_video_count: int = Field(default=0)
supports_first_last_frame: bool = Field(default=False, description="是否支持首尾帧模式")
supports_universal_reference: bool = Field(default=True, description="是否支持全能参考模式")
generate_url: str = Field(default="", max_length=512)
query_url: str = Field(default="", max_length=512)
is_active: bool = True
@@ -37,6 +39,8 @@ class VideoEnginePublic(BaseModel):
supported_durations: list[int] = []
max_image_count: int = 2
max_video_count: int = 0
supports_first_last_frame: bool = False
supports_universal_reference: bool = True
class VideoEngineListResponse(BaseModel):
video-gen-api/app/services/admin_credit_record_service.py
+12 -2
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from datetime import datetime
from datetime import datetime, timezone, timedelta
from typing import Any
from sqlalchemy import and_, case, distinct, func, or_, select
@@ -28,15 +28,25 @@ from app.models.shot_replicate_task_set import ShotReplicateTaskSet
from app.models.user import User
CST = timezone(timedelta(hours=8))
def _iso(dt: Any) -> str | None:
if dt is None:
return None
if isinstance(dt, datetime):
if dt.tzinfo is None:
# 数据库已经按东八区业务时间返回但丢了 tzinfo 时，不再额外 +8
return dt.replace(tzinfo=CST).isoformat()
return dt.astimezone(CST).isoformat()
try:
return dt.isoformat()
except Exception:
return str(dt)
def _round2(value: Any) -> float:
try:
return round(float(value or 0), 2)
video-gen-api/app/services/generation_ai_service.py
+3 -1
View File
@@ -203,6 +203,8 @@ async def list_generation_ai_engine_options(db: AsyncSession) -> GenerationAIEng
priority=engine.priority or 0,
max_image_count=engine.max_image_count,
max_video_count=engine.max_video_count,
supports_first_last_frame=engine.supports_first_last_frame,
supports_universal_reference=engine.supports_universal_reference,
)
for engine in video_result.scalars().all()
]
@@ -347,7 +349,7 @@ async def create_async_generation_task(db: AsyncSession, current_user: User, req
media_references=_json(refs) if refs else None,
credits_cost=round(media_billing.total_charged, 2),
idempotency_key=req.idempotency_key,
deadline_at=now + timedelta(minutes=settings.CHATAPI_ASYNC_VIDEO_DEADLINE_MINUTES),
deadline_at=now + timedelta(hours=settings.CHATAPI_ASYNC_VIDEO_FINAL_DEADLINE_HOURS),
)
db.add(task)
video-gen-api/app/services/generation_history_delete_service.py
+514
View File
@@ -0,0 +1,514 @@
from __future__ import annotations
import json
from datetime import datetime, timezone
from typing import Iterable, Sequence
from fastapi import HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.common import ModuleEventTypeEnum
from app.enums.generation_history import (
GenerationHistorySourceEnum,
get_generation_history_source_label,
normalize_generation_history_source,
MAX_BATCH_DELETE_COUNT,
)
from app.enums.generation_task import ChatGenerationTaskStatus, GenerationMode
from app.enums.shot_replicate import ShotSegmentAnalysisStatusEnum, ShotSegmentReplicateStatusEnum, ShotSplitStatusEnum
from app.models.chat_generation_task import ChatGenerationTask
from app.models.generation_record import GenerationRecord
from app.models.module_generation_project import ModuleGenerationProject
from app.models.module_generation_step import ModuleGenerationStep
from app.models.shot_replicate_segment import ShotReplicateSegment
from app.models.user import User
from app.schemas.generation_ai import GenerationAIHistoryBatchDeleteOut
from app.services.module_generation_flow_base_service import is_active_chat_generation_task
from app.services.module_generation_log_service import log_module_event_file
# from app.services.operation_log import log_operation
from app.services.resource_accounting_service import (
SOURCE_MODEL_CHAT_TASK,
SOURCE_MODEL_GENERATION_RECORD,
SOURCE_MODEL_SHOT_SEGMENT,
soft_delete_generation_record_resources,
soft_delete_resources_by_source,
)
COMPLETED_STATUS = ChatGenerationTaskStatus.COMPLETED.value
_ACTIVE_SPLIT_STATUSES = {
ShotSplitStatusEnum.PENDING.value,
ShotSplitStatusEnum.PROCESSING.value,
ShotSplitStatusEnum.RETRY_WAITING.value,
}
_ACTIVE_SEGMENT_ANALYSIS_STATUSES = {
ShotSegmentAnalysisStatusEnum.PENDING.value,
ShotSegmentAnalysisStatusEnum.PROCESSING.value,
}
_ACTIVE_SEGMENT_REPLICATE_STATUSES = {
ShotSegmentReplicateStatusEnum.PROCESSING.value,
}
def _normalize_ids(ids: Sequence[str] | Iterable[str]) -> list[str]:
normalized = [str(item).strip() for item in ids if str(item or "").strip()]
if not normalized:
raise HTTPException(status_code=400, detail="ids 不能为空")
if len(normalized) > MAX_BATCH_DELETE_COUNT:
raise HTTPException(status_code=400, detail=f"单次最多删除 {MAX_BATCH_DELETE_COUNT} 条记录")
if len(normalized) != len(set(normalized)):
raise HTTPException(status_code=400, detail="ids 不允许重复")
return normalized
def _missing_ids(request_ids: list[str], actual_ids: Iterable[str]) -> list[str]:
actual_set = {str(item) for item in actual_ids if item}
return [item for item in request_ids if item not in actual_set]
def _raise_missing_if_any(*, ids: list[str], found_ids: Iterable[str], message: str) -> None:
missing = _missing_ids(ids, found_ids)
if missing:
raise HTTPException(
status_code=404,
detail={
"message": message,
"missing_ids": missing,
"missing_count": len(missing),
},
)
def _raise_invalid_if_any(*, invalid_ids: list[str], message: str, status_code: int = 409) -> None:
if invalid_ids:
raise HTTPException(
status_code=status_code,
detail={
"message": message,
"invalid_ids": invalid_ids,
"invalid_count": len(invalid_ids),
},
)
def _task_id_list(steps: list[ModuleGenerationStep]) -> list[str]:
return list(dict.fromkeys(step.chat_task_id for step in steps if step.chat_task_id))
async def _load_chat_tasks_for_steps(
db: AsyncSession,
steps: list[ModuleGenerationStep],
) -> dict[str, ChatGenerationTask]:
task_ids = _task_id_list(steps)
if not task_ids:
return {}
result = await db.execute(
select(ChatGenerationTask)
.where(
ChatGenerationTask.id.in_(task_ids),
ChatGenerationTask.deleted_at.is_(None),
)
.with_for_update()
)
return {task.id: task for task in result.scalars().all()}
def _assert_no_active_chat_tasks(tasks: Iterable[ChatGenerationTask]) -> None:
active_task_ids = [task.id for task in tasks if is_active_chat_generation_task(task)]
if active_task_ids:
raise HTTPException(
status_code=409,
detail={
"message": "当前存在生成中任务，请等待生成完成或失败后再操作",
"active_count": len(active_task_ids),
},
)
# async def _log_history_batch_delete(
# db: AsyncSession,
# *,
# current_user: User,
# result: GenerationAIHistoryBatchDeleteOut,
# ) -> None:
# detail = {
# "history_source": result.history_source,
# "history_source_label": result.history_source_label,
# "requested_count": result.requested_count,
# "deleted_count": result.deleted_count,
# "requested_ids": result.requested_ids,
# "deleted_ids": result.deleted_ids,
# "generation_record_ids": result.generation_record_ids,
# "chat_task_ids": result.chat_task_ids,
# "module_project_ids": result.module_project_ids,
# "shot_segment_ids": result.shot_segment_ids,
# "freed_size_bytes": result.freed_size_bytes,
# }
# await log_operation(
# db,
# current_user.id,
# current_user.username,
# f"批量删除素材云历史-{result.history_source_label or result.history_source}",
# "DELETE",
# "/generation-ai/history/batch",
# detail=json.dumps(detail, ensure_ascii=False, default=str),
# )
def _build_out(
*,
source: GenerationHistorySourceEnum,
requested_ids: list[str],
deleted_ids: list[str],
generation_record_ids: list[str] | None = None,
chat_task_ids: list[str] | None = None,
module_project_ids: list[str] | None = None,
shot_segment_ids: list[str] | None = None,
freed_size_bytes: int = 0,
) -> GenerationAIHistoryBatchDeleteOut:
return GenerationAIHistoryBatchDeleteOut(
message="删除成功",
history_source=source.value,
history_source_label=get_generation_history_source_label(source),
requested_count=len(requested_ids),
deleted_count=len(deleted_ids),
requested_ids=requested_ids,
deleted_ids=deleted_ids,
generation_record_ids=generation_record_ids or [],
chat_task_ids=chat_task_ids or [],
module_project_ids=module_project_ids or [],
shot_segment_ids=shot_segment_ids or [],
deleted=True,
freed_size_bytes=int(freed_size_bytes or 0),
)
async def _delete_generation_records(
db: AsyncSession,
*,
current_user: User,
source: GenerationHistorySourceEnum,
ids: list[str],
deleted_at: datetime,
) -> GenerationAIHistoryBatchDeleteOut:
result = await db.execute(
select(GenerationRecord)
.where(
GenerationRecord.id.in_(ids),
GenerationRecord.user_id == current_user.id,
GenerationRecord.deleted_at.is_(None),
)
.with_for_update()
)
records = list(result.scalars().all())
_raise_missing_if_any(ids=ids, found_ids=[record.id for record in records], message="项目生成记录不存在或已删除")
invalid_ids = [
record.id
for record in records
if record.status != COMPLETED_STATUS or record.generated_at is None
]
_raise_invalid_if_any(invalid_ids=invalid_ids, message="项目生成记录只有生成完成后才能删除")
freed_size = await soft_delete_generation_record_resources(db, [record.id for record in records], deleted_at=deleted_at)
for record in records:
record.deleted_at = deleted_at
return _build_out(
source=source,
requested_ids=ids,
deleted_ids=ids,
generation_record_ids=ids,
freed_size_bytes=freed_size,
)
async def _delete_chat_tasks(
db: AsyncSession,
*,
current_user: User,
source: GenerationHistorySourceEnum,
ids: list[str],
deleted_at: datetime,
) -> GenerationAIHistoryBatchDeleteOut:
result = await db.execute(
select(ChatGenerationTask)
.where(
ChatGenerationTask.id.in_(ids),
ChatGenerationTask.user_id == current_user.id,
ChatGenerationTask.generation_mode == GenerationMode.CHATAPI_ASYNC.value,
ChatGenerationTask.deleted_at.is_(None),
)
.with_for_update()
)
tasks = list(result.scalars().all())
_raise_missing_if_any(ids=ids, found_ids=[task.id for task in tasks], message="AI 创作记录不存在或已删除")
invalid_ids = [
task.id
for task in tasks
if task.status != COMPLETED_STATUS or task.generated_at is None
]
_raise_invalid_if_any(invalid_ids=invalid_ids, message="AI 创作记录只有生成完成后才能删除")
freed_size = await soft_delete_resources_by_source(
db,
source_model=SOURCE_MODEL_CHAT_TASK,
source_ids=[task.id for task in tasks],
deleted_at=deleted_at,
)
for task in tasks:
task.deleted_at = deleted_at
return _build_out(
source=source,
requested_ids=ids,
deleted_ids=ids,
chat_task_ids=ids,
freed_size_bytes=freed_size,
)
async def _load_module_projects_by_ids(
db: AsyncSession,
*,
current_user: User,
source: GenerationHistorySourceEnum,
project_ids: list[str],
) -> list[ModuleGenerationProject]:
result = await db.execute(
select(ModuleGenerationProject)
.where(
ModuleGenerationProject.id.in_(project_ids),
ModuleGenerationProject.user_id == current_user.id,
ModuleGenerationProject.module == source.value,
ModuleGenerationProject.deleted_at.is_(None),
)
.with_for_update()
)
projects = list(result.scalars().all())
_raise_missing_if_any(ids=project_ids, found_ids=[project.id for project in projects], message="模块生成项目不存在或已删除")
return projects
async def _soft_delete_module_projects(
db: AsyncSession,
*,
source: GenerationHistorySourceEnum,
projects: list[ModuleGenerationProject],
deleted_at: datetime,
) -> tuple[list[str], list[str], int]:
project_ids = list(dict.fromkeys(project.id for project in projects))
if not project_ids:
return [], [], 0
step_result = await db.execute(
select(ModuleGenerationStep)
.where(
ModuleGenerationStep.project_id.in_(project_ids),
ModuleGenerationStep.module == source.value,
ModuleGenerationStep.deleted_at.is_(None),
ModuleGenerationStep.is_current == True,
)
.with_for_update()
)
steps = list(step_result.scalars().all())
task_map = await _load_chat_tasks_for_steps(db, steps)
_assert_no_active_chat_tasks(task_map.values())
chat_task_ids = list(task_map.keys())
freed_size = await soft_delete_resources_by_source(
db,
source_model=SOURCE_MODEL_CHAT_TASK,
source_ids=chat_task_ids,
deleted_at=deleted_at,
)
steps_by_project: dict[str, list[ModuleGenerationStep]] = {}
for step in steps:
steps_by_project.setdefault(step.project_id, []).append(step)
step.deleted_at = deleted_at
step.is_current = False
for task in task_map.values():
task.deleted_at = deleted_at
for project in projects:
project.deleted_at = deleted_at
project.final_image_url = None
project.final_video_url = None
project.final_video_cover_url = None
project.completed_at = None
project_steps = steps_by_project.get(project.id, [])
log_module_event_file(
module=source.value,
event_type=ModuleEventTypeEnum.PROJECT_DELETED.value,
project_id=project.id,
user_id=project.user_id,
message="素材云历史批量删除模块项目",
detail={
"source": "generation_history_batch_delete",
"step_ids": [step.id for step in project_steps],
"step_codes": [step.step_code for step in project_steps],
"chat_task_ids": [step.chat_task_id for step in project_steps if step.chat_task_id],
"refund_unfinished": False,
},
)
return project_ids, chat_task_ids, int(freed_size or 0)
async def _delete_hot_opening_projects(
db: AsyncSession,
*,
current_user: User,
source: GenerationHistorySourceEnum,
ids: list[str],
deleted_at: datetime,
) -> GenerationAIHistoryBatchDeleteOut:
projects = await _load_module_projects_by_ids(db, current_user=current_user, source=source, project_ids=ids)
module_project_ids, chat_task_ids, freed_size = await _soft_delete_module_projects(
db,
source=source,
projects=projects,
deleted_at=deleted_at,
)
return _build_out(
source=source,
requested_ids=ids,
deleted_ids=ids,
module_project_ids=module_project_ids,
chat_task_ids=chat_task_ids,
freed_size_bytes=freed_size,
)
def _assert_segments_not_active(segments: list[ShotReplicateSegment]) -> None:
invalid_ids = [
segment.id
for segment in segments
if segment.split_status in _ACTIVE_SPLIT_STATUSES
or segment.analysis_status in _ACTIVE_SEGMENT_ANALYSIS_STATUSES
or segment.replicate_status in _ACTIVE_SEGMENT_REPLICATE_STATUSES
]
_raise_invalid_if_any(invalid_ids=invalid_ids, message="拆镜片段仍在分割、分析或复刻处理中，暂不能删除")
async def _delete_shot_segments(
db: AsyncSession,
*,
current_user: User,
source: GenerationHistorySourceEnum,
ids: list[str],
deleted_at: datetime,
) -> GenerationAIHistoryBatchDeleteOut:
result = await db.execute(
select(ShotReplicateSegment)
.where(
ShotReplicateSegment.id.in_(ids),
ShotReplicateSegment.user_id == current_user.id,
ShotReplicateSegment.deleted_at.is_(None),
)
.with_for_update()
)
segments = list(result.scalars().all())
_raise_missing_if_any(ids=ids, found_ids=[segment.id for segment in segments], message="拆镜复刻片段不存在或已删除")
_assert_segments_not_active(segments)
missing_project_segment_ids = [segment.id for segment in segments if not segment.module_project_id]
_raise_invalid_if_any(invalid_ids=missing_project_segment_ids, message="拆镜复刻片段尚未关联复刻项目，不能按素材云历史删除")
module_project_ids = list(dict.fromkeys(segment.module_project_id for segment in segments if segment.module_project_id))
projects = await _load_module_projects_by_ids(
db,
current_user=current_user,
source=source,
project_ids=module_project_ids,
)
deleted_project_ids, chat_task_ids, project_freed_size = await _soft_delete_module_projects(
db,
source=source,
projects=projects,
deleted_at=deleted_at,
)
segment_freed_size = await soft_delete_resources_by_source(
db,
source_model=SOURCE_MODEL_SHOT_SEGMENT,
source_ids=[segment.id for segment in segments],
deleted_at=deleted_at,
)
for segment in segments:
segment.deleted_at = deleted_at
return _build_out(
source=source,
requested_ids=ids,
deleted_ids=ids,
module_project_ids=deleted_project_ids,
chat_task_ids=chat_task_ids,
shot_segment_ids=ids,
freed_size_bytes=int(project_freed_size or 0) + int(segment_freed_size or 0),
)
async def batch_delete_generation_history_items(
db: AsyncSession,
*,
current_user: User,
history_source: str,
ids: Sequence[str] | Iterable[str],
) -> GenerationAIHistoryBatchDeleteOut:
"""
按素材云 history_source 批量软删除历史记录。
事务由 get_db 统一提交/回滚；本服务只 flush，不主动 commit。
所有分支均为“先批量查询校验，再统一软删”，任何校验失败都会整体回滚。
"""
try:
source = normalize_generation_history_source(history_source)
except ValueError as exc:
raise HTTPException(status_code=400, detail="history_source 不支持") from exc
normalized_ids = _normalize_ids(ids)
deleted_at = datetime.now(timezone.utc)
if source == GenerationHistorySourceEnum.GENERATION_RECORD:
result = await _delete_generation_records(
db,
current_user=current_user,
source=source,
ids=normalized_ids,
deleted_at=deleted_at,
)
elif source == GenerationHistorySourceEnum.CHAT_TASK:
result = await _delete_chat_tasks(
db,
current_user=current_user,
source=source,
ids=normalized_ids,
deleted_at=deleted_at,
)
elif source == GenerationHistorySourceEnum.HOT_OPENING_REPLICATE:
result = await _delete_hot_opening_projects(
db,
current_user=current_user,
source=source,
ids=normalized_ids,
deleted_at=deleted_at,
)
elif source == GenerationHistorySourceEnum.SHOT_REPLICATE:
result = await _delete_shot_segments(
db,
current_user=current_user,
source=source,
ids=normalized_ids,
deleted_at=deleted_at,
)
else:
raise HTTPException(status_code=400, detail="history_source 不支持")
# await _log_history_batch_delete(db, current_user=current_user, result=result)
await db.flush()
return result
video-gen-api/app/services/generation_poll_schedule_service.py
+153
View File
@@ -0,0 +1,153 @@
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from app.config import settings
from app.enums.generation_task import GenerationType
from app.models.chat_generation_task import ChatGenerationTask
from app.services.redis_registry_service import ensure_aware_utc
@dataclass(slots=True)
class PollScheduleDecision:
delay_seconds: int
next_poll_at: datetime
poll_interval_seconds: int
direct_countdown: bool
final_poll: bool
reason: str
def utc_now() -> datetime:
return datetime.now(timezone.utc)
def is_video_generation_task(task: ChatGenerationTask) -> bool:
return str(getattr(task, "gen_type", "") or "").lower() == GenerationType.VIDEO.value
def video_final_deadline_from(now: datetime | None = None) -> datetime:
current_time = now or utc_now()
hours = max(1, int(settings.CHATAPI_ASYNC_VIDEO_FINAL_DEADLINE_HOURS or 24))
return current_time + timedelta(hours=hours)
def ensure_video_poll_fields(task: ChatGenerationTask, *, now: datetime | None = None) -> None:
"""补齐视频轮询调度字段，兼容历史任务。"""
if not is_video_generation_task(task):
return
current_time = now or utc_now()
if ensure_aware_utc(getattr(task, "poll_started_at", None)) is None:
task.poll_started_at = current_time
if ensure_aware_utc(getattr(task, "deadline_at", None)) is None:
task.deadline_at = video_final_deadline_from(current_time)
if getattr(task, "poll_interval_seconds", None) is None:
task.poll_interval_seconds = 0
def is_final_poll_due(task: ChatGenerationTask, *, now: datetime | None = None) -> bool:
deadline_at = ensure_aware_utc(getattr(task, "deadline_at", None))
return bool(deadline_at and deadline_at <= (now or utc_now()))
def is_poll_not_due(task: ChatGenerationTask, *, now: datetime | None = None, tolerance_seconds: int = 1) -> bool:
"""判断当前 poll 任务是否早于 next_poll_at。只对视频降频轮询生效。"""
if not is_video_generation_task(task):
return False
if is_final_poll_due(task, now=now):
return False
next_poll_at = ensure_aware_utc(getattr(task, "next_poll_at", None))
if next_poll_at is None:
return False
return next_poll_at > (now or utc_now()) + timedelta(seconds=max(0, int(tolerance_seconds)))
def _clamp_positive_seconds(value: int | float | None, default: int) -> int:
try:
parsed = int(value if value is not None else default)
except (TypeError, ValueError):
parsed = default
return max(1, parsed)
def build_video_pending_poll_schedule(
task: ChatGenerationTask,
*,
now: datetime | None = None,
) -> PollScheduleDecision:
"""计算视频任务 pending/running 后的下一次轮询时间。"""
current_time = now or utc_now()
ensure_video_poll_fields(task, now=current_time)
deadline_at = ensure_aware_utc(task.deadline_at)
if deadline_at and deadline_at <= current_time:
return PollScheduleDecision(
delay_seconds=0,
next_poll_at=current_time,
poll_interval_seconds=int(task.poll_interval_seconds or 0),
direct_countdown=True,
final_poll=True,
reason="video_final_poll_due",
)
poll_started_at = ensure_aware_utc(task.poll_started_at) or current_time
elapsed_seconds = max(0, int((current_time - poll_started_at).total_seconds()))
high_freq_seconds = max(0, int(settings.CHATAPI_ASYNC_VIDEO_HIGH_FREQ_MINUTES or 10)) * 60
high_freq_poll_seconds = _clamp_positive_seconds(settings.CHATAPI_ASYNC_VIDEO_HIGH_FREQ_POLL_SECONDS, 30)
initial_backoff_seconds = _clamp_positive_seconds(settings.CHATAPI_ASYNC_VIDEO_BACKOFF_INITIAL_SECONDS, 60)
multiplier = max(1, int(settings.CHATAPI_ASYNC_VIDEO_BACKOFF_MULTIPLIER or 2))
max_backoff_seconds = _clamp_positive_seconds(settings.CHATAPI_ASYNC_VIDEO_BACKOFF_MAX_SECONDS, 3600)
if elapsed_seconds < high_freq_seconds:
delay_seconds = high_freq_poll_seconds
interval_seconds = int(task.poll_interval_seconds or 0)
reason = "video_high_freq_poll"
else:
previous_interval = int(task.poll_interval_seconds or 0)
if previous_interval < initial_backoff_seconds:
interval_seconds = initial_backoff_seconds
else:
interval_seconds = min(previous_interval * multiplier, max_backoff_seconds)
delay_seconds = interval_seconds
reason = "video_backoff_poll"
next_poll_at = current_time + timedelta(seconds=delay_seconds)
final_poll = False
if deadline_at and next_poll_at >= deadline_at:
next_poll_at = deadline_at
delay_seconds = max(0, int((deadline_at - current_time).total_seconds()))
final_poll = delay_seconds <= 0
reason = "video_schedule_to_final_deadline"
direct_max = max(0, int(settings.CHATAPI_ASYNC_VIDEO_DIRECT_COUNTDOWN_MAX_SECONDS or 300))
return PollScheduleDecision(
delay_seconds=delay_seconds,
next_poll_at=next_poll_at,
poll_interval_seconds=interval_seconds,
direct_countdown=delay_seconds <= direct_max,
final_poll=final_poll,
reason=reason,
)
def build_default_poll_schedule(
task: ChatGenerationTask,
*,
now: datetime | None = None,
delay_seconds: int | None = None,
reason: str = "default_poll",
) -> PollScheduleDecision:
"""图片和旧逻辑兼容用的固定间隔轮询计划。"""
current_time = now or utc_now()
delay = _clamp_positive_seconds(delay_seconds, int(settings.CHATAPI_ASYNC_POLL_INTERVAL_SECONDS or 30))
next_poll_at = current_time + timedelta(seconds=delay)
return PollScheduleDecision(
delay_seconds=delay,
next_poll_at=next_poll_at,
poll_interval_seconds=int(getattr(task, "poll_interval_seconds", 0) or 0),
direct_countdown=True,
final_poll=False,
reason=reason,
)
video-gen-api/app/services/generation_recovery_service.py
+163 -13
View File
@@ -9,11 +9,13 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.config import settings
from app.enums.celery_queue import CeleryQueue
from app.enums.generation_task import (
ALLOWED_GENERATION_MODES,
ChatGenerationPipelineStage,
ChatGenerationTaskEventType,
ChatGenerationTaskStatus,
GenerationType,
)
from app.models.chat_generation_task import ChatGenerationTask
from app.services.celery_download_recovery_service import (
@@ -25,6 +27,7 @@ from app.services.celery_download_recovery_service import (
)
from app.services.generation_log_service import log_task_event
from app.services.generation_module_hook_service import notify_chat_generation_task_finished
from app.services.generation_poll_schedule_service import ensure_video_poll_fields, is_poll_not_due, is_video_generation_task
from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once
from app.services.redis_registry_service import (
redis_get_due_registry_ids,
@@ -35,7 +38,7 @@ from app.services.redis_registry_service import (
logger = logging.getLogger("video_gen")
POLL_QUEUE = "gen_provider_poll"
POLL_QUEUE = CeleryQueue.GEN_PROVIDER_POLL.value
def _now() -> datetime:
@@ -342,7 +345,7 @@ async def _mark_timeout(
await _remove_poll_active(task.id)
await log_task_event(
task,
event_type="TASK_TIMEOUT",
event_type=ChatGenerationTaskEventType.TASK_TIMEOUT.value,
to_status="failed",
to_stage=ChatGenerationPipelineStage.TIMEOUT.value,
)
@@ -414,7 +417,7 @@ async def recover_one_generation_task(
await _remove_poll_active(task.id)
await log_task_event(
task,
event_type="GENERATION_RECOVERY_ENQUEUE",
event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value,
message=f"{source} 发现任务已存在 remote_result_url，恢复投递下载队列",
detail={
"pipeline_stage": task.pipeline_stage,
@@ -439,7 +442,7 @@ async def recover_one_generation_task(
await db.commit()
await log_task_event(
task,
event_type="GENERATION_RECOVERY_ENQUEUE",
event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value,
message=f"{source} 发现任务已到 deadline 且存在供应商任务ID，投递 poll 队列做最终查询",
detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload},
)
@@ -453,26 +456,51 @@ async def recover_one_generation_task(
await log_task_event(
task,
event_type="GENERATION_RECOVERY_TIMEOUT",
event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_TIMEOUT.value,
message=f"{source} 发现任务已到 deadline，且没有 remote_result_url/供应商任务ID，按超时失败处理",
detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload},
)
return await _mark_timeout(db, task)
# 未过 deadline：有供应商任务 ID 才允许恢复到 poll 队列。
# 视频任务如果 next_poll_at 未到期，不提前 poll，只刷新 active 注册表等待 Beat dispatcher 到期投递。
if has_provider_task_id:
if is_video_generation_task(task):
ensure_video_poll_fields(task, now=current_time)
if is_poll_not_due(task, now=current_time):
await db.commit()
await register_poll_active(
task,
check_at=task.next_poll_at,
next_poll_at=task.next_poll_at,
reason=f"{source}_video_poll_not_due",
)
await log_task_event(
task,
event_type=ChatGenerationTaskEventType.POLL_SKIP_NOT_DUE.value,
message=f"{source} 发现视频任务尚未到下一次轮询时间，启动容灾不提前投递 poll",
detail={
"pipeline_stage": task.pipeline_stage,
"payload": redis_payload,
"next_poll_at": task.next_poll_at,
},
)
return "skip_video_poll_not_due"
task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value
task.next_poll_at = _poll_queue_timeout_at(current_time)
await db.commit()
await log_task_event(
task,
event_type="GENERATION_RECOVERY_ENQUEUE",
event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value,
message=f"{source} 发现任务存在供应商任务ID，恢复投递轮询队列",
detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload},
)
poll_generation_task.apply_async(args=[task.id], queue=POLL_QUEUE, countdown=0)
await register_poll_active(
task,
check_at=_poll_queue_timeout_at(),
check_at=task.next_poll_at,
next_poll_at=task.next_poll_at,
reason=f"{source}_has_provider_task_id",
)
return "recover_poll_has_provider_id"
@@ -499,13 +527,13 @@ async def recover_one_generation_task(
await _remove_poll_active(task.id)
await log_task_event(
task,
event_type="GENERATION_RECOVERY_ENQUEUE",
event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value,
message=f"{source} 发现任务未超时且缺少 remote_result_url/供应商任务ID，恢复投递创建队列",
detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload},
)
chatapi_create_generation_task.apply_async(
args=[task.id],
queue="gen_chatapi_create",
queue=CeleryQueue.GEN_CHATAPI_CREATE.value,
countdown=0,
)
return "recover_create_no_remote_no_provider_before_deadline"
@@ -517,13 +545,13 @@ async def recover_one_generation_task(
await _remove_poll_active(task.id)
await log_task_event(
task,
event_type="GENERATION_RECOVERY_ENQUEUE",
event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value,
message=f"{source} 发现 result_ready 但缺少 remote_result_url，未超时，恢复投递创建队列",
detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload},
)
chatapi_create_generation_task.apply_async(
args=[task.id],
queue="gen_chatapi_create",
queue=CeleryQueue.GEN_CHATAPI_CREATE.value,
countdown=0,
)
return "recover_create_result_ready_no_url_before_deadline"
@@ -534,7 +562,7 @@ async def recover_one_generation_task(
async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
"""启动时生成链路容灾扫描。
不新增 Celery beat，不新增 worker 命令；worker 启动时由 Redis 锁保证只投递一次。
启动容灾由 worker_ready 触发，只跑一次完整恢复；周期性视频到期轮询由 Celery Beat 调度 dispatch_due_poll_tasks。
恢复顺序：
1. Redis poll active_index 到期任务；
2. DB fallback 扫描 queued/creating/waiting_remote/polling/result_ready；
@@ -630,4 +658,126 @@ async def recover_generation_tasks_once(db: AsyncSession) -> dict[str, Any]:
"checked": len(checked_ids),
"db_checked": total_db_checked,
"results": results,
}
}
async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]:
"""周期性轻量到期轮询调度。
只处理视频任务的 next_poll_at 到期记录，不替代启动容灾 recover_generation_tasks_once。
Beat 每分钟触发本任务，本任务把真实供应商轮询投递到 gen_provider_poll 队列。
"""
from app.tasks.generation_poll_tasks import poll_generation_task, register_poll_active
current_time = _now()
batch_size = max(1, int(settings.POLL_DUE_DISPATCH_BATCH_SIZE or 100))
poll_lease_expired_at = current_time - timedelta(seconds=int(settings.POLL_TASK_LEASE_SECONDS or 300))
query_result = await db.execute(
select(ChatGenerationTask)
.where(
ChatGenerationTask.deleted_at.is_(None),
ChatGenerationTask.generation_mode.in_(list(ALLOWED_GENERATION_MODES)),
ChatGenerationTask.status == ChatGenerationTaskStatus.GENERATING.value,
ChatGenerationTask.gen_type == GenerationType.VIDEO.value,
ChatGenerationTask.next_poll_at.is_not(None),
ChatGenerationTask.next_poll_at <= current_time,
ChatGenerationTask.pipeline_stage.in_(
[
ChatGenerationPipelineStage.WAITING_REMOTE.value,
ChatGenerationPipelineStage.POLLING.value,
]
),
)
.order_by(ChatGenerationTask.next_poll_at.asc(), ChatGenerationTask.updated_at.asc())
.limit(batch_size)
.with_for_update(skip_locked=True)
)
tasks = query_result.scalars().all()
results: dict[str, int] = {}
dispatched_task_ids: list[str] = []
for task in tasks:
action = "skip_unknown"
try:
if task.pipeline_stage == ChatGenerationPipelineStage.POLLING.value:
last_poll_at = ensure_aware_utc(task.last_poll_at)
if last_poll_at and last_poll_at > poll_lease_expired_at:
await register_poll_active(
task,
check_at=_poll_queue_timeout_at(last_poll_at),
next_poll_at=task.next_poll_at,
reason="due_dispatch_polling_lease_alive",
)
action = "skip_polling_lease_alive"
continue
if not (task.seedance_task_id or task.provider_task_id):
# dispatcher 不负责重新 create；没有 provider id 的异常状态交给启动容灾或 create 任务处理。
await log_task_event(
task,
event_type=ChatGenerationTaskEventType.POLL_DISPATCH_SKIP.value,
message="视频到期轮询调度跳过：缺少外部任务ID",
detail={"pipeline_stage": task.pipeline_stage, "next_poll_at": task.next_poll_at},
)
action = "skip_no_provider_task_id"
continue
task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value
# 设置一个队列消费保护时间，避免 Beat 下一分钟看到旧 next_poll_at 又重复投递。
task.next_poll_at = _poll_queue_timeout_at(current_time)
dispatched_task_ids.append(task.id)
action = "dispatch_poll"
except Exception as exc:
logger.exception("视频到期轮询调度单条处理失败。task_id=%s", getattr(task, "id", None))
action = "error"
await log_task_event(
task,
event_type=ChatGenerationTaskEventType.POLL_DISPATCH_SKIP.value,
message=f"视频到期轮询调度单条处理失败：{exc}",
)
finally:
results[action] = results.get(action, 0) + 1
await db.commit()
fresh_tasks = []
if dispatched_task_ids:
fresh_result = await db.execute(
select(ChatGenerationTask)
.where(
ChatGenerationTask.id.in_(dispatched_task_ids),
ChatGenerationTask.deleted_at.is_(None),
)
.execution_options(populate_existing=True)
)
fresh_tasks = fresh_result.scalars().all()
enqueued_count = 0
for task in fresh_tasks:
await register_poll_active(
task,
check_at=task.next_poll_at or _poll_queue_timeout_at(current_time),
next_poll_at=task.next_poll_at,
reason="due_dispatch_poll_queued",
)
await log_task_event(
task,
event_type=ChatGenerationTaskEventType.POLL_DISPATCH_DUE.value,
message="视频 next_poll_at 到期，已投递 provider poll 队列",
detail={"next_poll_at": task.next_poll_at, "queue": POLL_QUEUE},
)
poll_generation_task.apply_async(args=[task.id], queue=POLL_QUEUE, countdown=0)
enqueued_count += 1
# 如果 log_task_event 内部不 commit，这里要提交一次
if fresh_tasks:
await db.commit()
return {
"checked": len(tasks),
"dispatched": len(dispatched_task_ids),
"enqueued": enqueued_count,
"results": results,
}
video-gen-api/app/services/generation_task_factory_service.py
+1 -1
View File
@@ -195,7 +195,7 @@ async def create_chat_generation_task_for_module(
media_references=_json(refs) if refs else None,
credits_cost=round(media_billing.total_charged, 2),
idempotency_key=backend_idempotency_key,
deadline_at=now + timedelta(minutes=settings.CHATAPI_ASYNC_VIDEO_DEADLINE_MINUTES),
deadline_at=now + timedelta(hours=settings.CHATAPI_ASYNC_VIDEO_FINAL_DEADLINE_HOURS),
)
db.add(task)
video-gen-api/app/services/module_generation_flow_base_service.py
+151 -26
View File
@@ -9,14 +9,17 @@ from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.enums.common import ModuleEventTypeEnum, ModuleStepStatusEnum
from app.enums.generation_task import ChatGenerationPipelineStage, ChatGenerationTaskStatus
from app.models.chat_generation_task import ChatGenerationTask
from app.models.module_generation_project import ModuleGenerationProject
from app.models.module_generation_step import ModuleGenerationStep
from app.models.user import User
from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once
from app.enums.module_generation_flow import ModuleGenerationFlowConfig
from app.services.module_generation_step_common_service import build_step_input, build_step_output, utc_now
from app.services.resource_accounting_service import soft_delete_chat_task_resources
from app.services.resource_accounting_service import (
SOURCE_MODEL_CHAT_TASK,
soft_delete_resources_by_source,
)
from app.utils.id_gen import generate_id
LogModuleEventCallable = Callable[..., Awaitable[None]]
@@ -224,6 +227,106 @@ def _clear_project_final_resources_by_deleted_steps(
return cleared_fields
ACTIVE_CHAT_TASK_BLOCK_STATUSES = {
ChatGenerationTaskStatus.PENDING.value,
ChatGenerationTaskStatus.GENERATING.value,
}
ACTIVE_CHAT_TASK_BLOCK_STAGES = {
ChatGenerationPipelineStage.QUEUED.value,
ChatGenerationPipelineStage.PREPARING.value,
ChatGenerationPipelineStage.CREATING_PROVIDER_TASK.value,
ChatGenerationPipelineStage.WAITING_REMOTE.value,
ChatGenerationPipelineStage.POLLING.value,
ChatGenerationPipelineStage.RESULT_READY.value,
ChatGenerationPipelineStage.DOWNLOAD_QUEUED.value,
ChatGenerationPipelineStage.DOWNLOADING.value,
ChatGenerationPipelineStage.RETRY_WAITING.value,
}
def is_active_chat_generation_task(task: ChatGenerationTask) -> bool:
"""判断 ChatGenerationTask 是否仍处于不可主动删除/废弃的处理中状态。"""
return bool(
task.status in ACTIVE_CHAT_TASK_BLOCK_STATUSES
or (task.pipeline_stage in ACTIVE_CHAT_TASK_BLOCK_STAGES)
)
def _unique_ids(values: list[str | None] | tuple[str | None, ...]) -> list[str]:
return list(dict.fromkeys(str(value) for value in values if value))
async def load_chat_tasks_for_steps(
db: AsyncSession,
steps: list[ModuleGenerationStep],
*,
for_update: bool = False,
) -> dict[str, ChatGenerationTask]:
ids = _unique_ids([step.chat_task_id for step in steps])
if not ids:
return {}
stmt = select(ChatGenerationTask).where(
ChatGenerationTask.id.in_(ids),
ChatGenerationTask.deleted_at.is_(None),
)
if for_update:
stmt = stmt.with_for_update()
result = await db.execute(stmt)
return {task.id: task for task in result.scalars().all()}
async def assert_no_active_chat_tasks_for_steps(
db: AsyncSession,
steps: list[ModuleGenerationStep],
*,
detail_message: str = "当前存在生成中任务，请等待生成完成或失败后再操作",
for_update: bool = True,
) -> dict[str, ChatGenerationTask]:
"""批量校验步骤关联任务是否有进行中任务。
用户主动修改上游步骤、删除项目、删除历史时，统一采用“生成中拦截、不退款”。
返回已批量加载的未删除 ChatGenerationTask，调用方可继续复用，避免重复查询。
"""
task_map = await load_chat_tasks_for_steps(db, steps, for_update=for_update)
active_task_ids = [task.id for task in task_map.values() if is_active_chat_generation_task(task)]
if active_task_ids:
raise HTTPException(
status_code=409,
detail={
"message": detail_message,
"active_count": len(active_task_ids),
},
)
return task_map
async def assert_project_has_no_active_chat_tasks(
db: AsyncSession,
*,
project: ModuleGenerationProject,
config: ModuleGenerationFlowConfig,
detail_message: str = "当前存在生成中任务，请等待生成完成或失败后再操作",
) -> dict[str, ChatGenerationTask]:
result = await db.execute(
select(ModuleGenerationStep)
.where(
ModuleGenerationStep.project_id == project.id,
ModuleGenerationStep.module == config.module,
ModuleGenerationStep.deleted_at.is_(None),
ModuleGenerationStep.chat_task_id.is_not(None),
)
.with_for_update()
)
steps = list(result.scalars().all())
return await assert_no_active_chat_tasks_for_steps(
db,
steps,
detail_message=detail_message,
for_update=True,
)
async def soft_delete_steps_from_index(
db: AsyncSession,
*,
@@ -232,7 +335,17 @@ async def soft_delete_steps_from_index(
config: ModuleGenerationFlowConfig,
log_module_event: LogModuleEventCallable,
deleted_at: datetime | None = None,
refund_unfinished: bool = False,
block_active_tasks: bool = True,
release_stats: dict[str, int] | None = None,
) -> list[ModuleGenerationStep]:
"""软删指定步骤及其后续当前版本步骤。
说明：
- 兼容旧调用方保留 refund_unfinished 参数，但用户主动修改/删除链路不再退款。
- 存在 pending/generating/下载中/轮询中等任务时直接 409 拦截。
- 已完成任务只做软删任务与 generated_resources，释放容量统计；失败任务只软删任务。
"""
deleted_at = deleted_at or utc_now()
result = await db.execute(
select(ModuleGenerationStep)
@@ -246,6 +359,21 @@ async def soft_delete_steps_from_index(
.with_for_update()
)
steps = list(result.scalars().all())
if not steps:
return []
task_map = await load_chat_tasks_for_steps(db, steps, for_update=True)
if block_active_tasks:
active_task_ids = [task.id for task in task_map.values() if is_active_chat_generation_task(task)]
if active_task_ids:
raise HTTPException(
status_code=409,
detail={
"message": "当前存在生成中任务，请等待生成完成或失败后再操作",
"active_count": len(active_task_ids),
},
)
deleted_step_codes = {step.step_code for step in steps}
cleared_project_fields = _clear_project_final_resources_by_deleted_steps(
project,
@@ -253,28 +381,27 @@ async def soft_delete_steps_from_index(
config=config,
)
completed_task_ids = [
task.id
for task in task_map.values()
if task.status == ChatGenerationTaskStatus.COMPLETED.value
]
if completed_task_ids:
released_size = await soft_delete_resources_by_source(
db,
source_model=SOURCE_MODEL_CHAT_TASK,
source_ids=completed_task_ids,
deleted_at=deleted_at,
)
if release_stats is not None:
release_stats["released_size_bytes"] = int(release_stats.get("released_size_bytes", 0)) + int(released_size or 0)
for step in steps:
step.is_current = False
step.deleted_at = deleted_at
if step.chat_task_id:
chat_result = await db.execute(
select(ChatGenerationTask)
.where(ChatGenerationTask.id == step.chat_task_id, ChatGenerationTask.deleted_at.is_(None))
.with_for_update()
.limit(1)
)
chat_task = chat_result.scalar_one_or_none()
if chat_task:
if chat_task.status == "completed":
await soft_delete_chat_task_resources(db, chat_task.id, deleted_at=deleted_at)
elif chat_task.status != "failed":
await mark_chat_generation_task_failed_and_refund_once(
db,
task=chat_task,
error_message=config.cancel_chat_task_error_message,
pipeline_stage="failed",
)
chat_task.deleted_at = deleted_at
if step.chat_task_id and step.chat_task_id in task_map:
task_map[step.chat_task_id].deleted_at = deleted_at
if steps:
await log_module_event(
db,
@@ -285,14 +412,12 @@ async def soft_delete_steps_from_index(
"step_ids": [step.id for step in steps],
"step_codes": [step.step_code for step in steps],
"cleared_project_fields": cleared_project_fields,
"block_active_tasks": block_active_tasks,
"refund_unfinished": False,
},
)
return steps
async def chat_tasks_by_id(db: AsyncSession, steps: list[ModuleGenerationStep]) -> dict[str, ChatGenerationTask]:
ids = [step.chat_task_id for step in steps if step.chat_task_id]
if not ids:
return {}
result = await db.execute(select(ChatGenerationTask).where(ChatGenerationTask.id.in_(ids)))
return {task.id: task for task in result.scalars().all()}
return await load_chat_tasks_for_steps(db, steps, for_update=False)
video-gen-api/app/services/resource_accounting_service.py
+1
View File
@@ -18,6 +18,7 @@ from app.utils.id_gen import generate_id
SOURCE_MODEL_CHAT_TASK = "ChatGenerationTask"
SOURCE_MODEL_GENERATION_RECORD = "GenerationRecord"
SOURCE_MODEL_SHOT_SEGMENT = "ShotReplicateSegment"
@dataclass(slots=True)
video-gen-api/app/services/shot_replicate_flow_service.py
+56 -4
View File
@@ -54,6 +54,7 @@ from app.services.module_generation_log_service import log_module_error, log_mod
from app.services.llm import optimize_prompt
from app.services.resource_accounting_service import soft_delete_chat_task_resources
from app.services.module_generation_flow_base_service import (
assert_project_has_no_active_chat_tasks as _base_assert_project_has_no_active_chat_tasks,
chat_tasks_by_id as _base_chat_tasks_by_id,
create_module_step as _base_create_step,
get_current_step_by_code as _base_get_current_step_by_code,
@@ -335,6 +336,8 @@ async def _soft_delete_steps_from_index(
project: ModuleGenerationProject,
start_index: int,
deleted_at: datetime | None = None,
refund_unfinished: bool = False,
release_stats: dict[str, int] | None = None,
) -> None:
await _base_soft_delete_steps_from_index(
db,
@@ -343,6 +346,8 @@ async def _soft_delete_steps_from_index(
config=FLOW_CONFIG,
log_module_event=log_module_event,
deleted_at=deleted_at,
refund_unfinished=refund_unfinished,
release_stats=release_stats,
)
@@ -1406,6 +1411,21 @@ async def handle_chat_generation_task_failed(db: AsyncSession, task: ChatGenerat
await log_module_event(db, project=project, step=step, event_type=ModuleEventTypeEnum.CHAT_TASK_FAILED.value, message=project.error_message, detail={"chat_task_id": task.id})
async def _assert_project_has_no_active_chat_tasks_for_delete(
db: AsyncSession,
*,
project: ModuleGenerationProject,
) -> None:
"""用户主动删除项目/切片时不退款；如仍有异步生成任务进行中，直接拦截。"""
await _base_assert_project_has_no_active_chat_tasks(
db,
project=project,
config=FLOW_CONFIG,
detail_message="当前拆镜复刻项目仍有生成中任务，暂不能删除",
)
async def mark_shot_replicate_step_dispatch_failed(
db: AsyncSession,
*,
@@ -1462,13 +1482,45 @@ async def mark_shot_replicate_step_dispatch_failed(
)
async def delete_shot_replicate_project(db: AsyncSession, *, current_user: User, project_id: str) -> ShotReplicateDeleteOut:
async def delete_shot_replicate_project(
db: AsyncSession,
*,
current_user: User,
project_id: str,
refund_unfinished: bool = False,
) -> ShotReplicateDeleteOut:
project = await _get_project_for_user(db, project_id=project_id, user=current_user, for_update=True)
deleted_at = _now()
release_stats: dict[str, int] = {"released_size_bytes": 0}
if not refund_unfinished:
await _assert_project_has_no_active_chat_tasks_for_delete(db, project=project)
project.deleted_at = deleted_at
await _soft_delete_steps_from_index(db, project=project, start_index=1, deleted_at=deleted_at)
await log_module_event(db, project=project, event_type=ModuleEventTypeEnum.PROJECT_DELETED.value, message="软删除拆镜复刻项目")
return ShotReplicateDeleteOut(message="项目已删除", project_id=project.id, deleted=True)
await _soft_delete_steps_from_index(
db,
project=project,
start_index=1,
deleted_at=deleted_at,
refund_unfinished=refund_unfinished,
release_stats=release_stats,
)
await log_module_event(
db,
project=project,
event_type=ModuleEventTypeEnum.PROJECT_DELETED.value,
message="软删除拆镜复刻项目",
detail={
"refund_unfinished": refund_unfinished,
"released_size_bytes": int(release_stats.get("released_size_bytes", 0)),
},
)
return ShotReplicateDeleteOut(
message="项目已删除",
project_id=project.id,
deleted=True,
released_size_bytes=int(release_stats.get("released_size_bytes", 0)),
)
async def create_shot_replicate_project_from_segment(
video-gen-api/app/services/shot_replicate_taskset_service.py
+101
View File
@@ -25,6 +25,7 @@ from app.models.shot_replicate_task_set import ShotReplicateTaskSet
from app.models.user import User
from app.schemas.shot_replicate import (
ShotAISuggestionOut,
ShotSegmentDeleteOut,
ShotSegmentDetailOut,
ShotSegmentListOut,
ShotSegmentOut,
@@ -38,6 +39,7 @@ from app.schemas.shot_replicate import (
ShotTaskSetOut,
)
from app.services.module_generation_log_service import log_module_event_file
from app.services.resource_accounting_service import SOURCE_MODEL_SHOT_SEGMENT, soft_delete_resources_by_source
from app.services.upload_video_asset_service import (
build_time_node,
validate_split_range,
@@ -615,3 +617,102 @@ async def segment_detail(db: AsyncSession, *, current_user: User, segment_id: st
project_result = await db.execute(select(ModuleGenerationProject).where(ModuleGenerationProject.id == segment.module_project_id).limit(1))
project = project_result.scalar_one_or_none()
return _segment_to_detail_out(segment, project)
async def delete_segment(
db: AsyncSession,
*,
current_user: User,
segment_id: str,
) -> ShotSegmentDeleteOut:
"""软删除拆镜片段。
只释放用户容量账本记录，不删除 segment_video_path 指向的物理文件。
如果片段已创建复刻项目，则联动调用项目删除逻辑，但用户主动删除不退款；
项目仍有生成中任务时会拒绝删除，避免异步任务继续写回软删数据。
"""
query = select(ShotReplicateSegment).where(ShotReplicateSegment.id == segment_id)
if not current_user.is_admin:
query = query.where(ShotReplicateSegment.user_id == current_user.id)
result = await db.execute(query.with_for_update().limit(1))
segment = result.scalar_one_or_none()
if not segment:
raise HTTPException(status_code=404, detail="拆镜片段不存在")
task_set_id = segment.task_set_id
module_project_id = segment.module_project_id
if segment.deleted_at is not None:
return ShotSegmentDeleteOut(
message="拆镜片段已删除",
segment_id=segment.id,
task_set_id=task_set_id,
deleted=True,
deleted_module_project_id=module_project_id,
released_size_bytes=0,
)
if segment.split_status == ShotSplitStatusEnum.PROCESSING.value:
raise HTTPException(status_code=400, detail="当前拆镜片段正在切割处理中，暂不能删除")
if segment.analysis_status == ShotSegmentAnalysisStatusEnum.PROCESSING.value:
raise HTTPException(status_code=400, detail="当前拆镜片段正在分析处理中，暂不能删除")
if segment.replicate_status == ShotSegmentReplicateStatusEnum.PROCESSING.value:
raise HTTPException(status_code=400, detail="当前拆镜片段关联的复刻流程正在处理中，暂不能删除")
deleted_at = _now()
released_size_bytes = await soft_delete_resources_by_source(
db,
source_model=SOURCE_MODEL_SHOT_SEGMENT,
source_ids=[segment.id],
deleted_at=deleted_at,
)
deleted_module_project_id: str | None = None
if module_project_id:
from app.services.shot_replicate_flow_service import delete_shot_replicate_project
project_delete_out = await delete_shot_replicate_project(
db,
current_user=current_user,
project_id=module_project_id,
refund_unfinished=False,
)
deleted_module_project_id = project_delete_out.project_id
released_size_bytes += int(project_delete_out.released_size_bytes or 0)
segment.deleted_at = deleted_at
segment.replicate_status = (
ShotSegmentReplicateStatusEnum.NOT_STARTED.value
if not deleted_module_project_id
else ShotSegmentReplicateStatusEnum.FAILED.value
)
await refresh_task_set_split_summary(db, task_set_id)
await db.flush()
log_module_event_file(
module=MODULE,
event_type="SHOT_SEGMENT_DELETED",
project_id=task_set_id,
step_id=segment.id,
user_id=segment.user_id,
message="软删除拆镜片段并释放用户容量账本记录",
detail={
"segment_id": segment.id,
"task_set_id": task_set_id,
"module_project_id": module_project_id,
"deleted_module_project_id": deleted_module_project_id,
"released_size_bytes": released_size_bytes,
"physical_file_deleted": False,
"refund": False,
},
)
return ShotSegmentDeleteOut(
message="拆镜片段已删除",
segment_id=segment.id,
task_set_id=task_set_id,
deleted=True,
deleted_module_project_id=deleted_module_project_id,
released_size_bytes=int(released_size_bytes or 0),
)
video-gen-api/app/services/user_oauth_service.py
+55 -4
View File
@@ -9,6 +9,7 @@ from app.config import settings
from app.models.user_oauth import UserOAuth
from app.models.user_oauth_app import UserOAuthApp
from app.utils.id_gen import generate_id
from app.tasks.token_refresh_task import _update_redis_token
#随机获取一个可用的应用配置
async def get_available_app(open_type: int, db: AsyncSession) -> dict:
@@ -172,10 +173,13 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int,
)
oauth = oauth.scalar_one_or_none()
if not oauth:
new_oauth_ids = []
for account in account_list:
oauth_id = generate_id()
new_oauth_ids.append(oauth_id)
#新增授权记录
db.add(UserOAuth(
id=generate_id(),
id=oauth_id,
account_id = str(account.get("account_id", "")),
account_name = account.get("account_name", ""),
account_role = account.get("account_role", ""),
@@ -191,7 +195,11 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int,
refresh_token_expired = refresh_token_expires_in,
material_auth_status = material_auth_status,
))
await db.commit()
await db.commit()
for oauth_id in new_oauth_ids:
await _update_redis_token(oauth_id, access_token, expires_in)
else:
# 查询现有授权记录（未删除的）
existing_accounts = await db.execute(
@@ -209,6 +217,9 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int,
new_account_ids = {str(account.get("account_id")) for account in account_list}
old_account_ids = set(existing_accounts.keys())
# 需要更新Redis的oauth_id列表
update_redis_ids = []
# 1. 软删除已消失的账户
for account_id in old_account_ids - new_account_ids:
existing_accounts[account_id].deleted_at = datetime.now()
@@ -226,10 +237,13 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int,
existing_oauth.refresh_token = refresh_token
existing_oauth.refresh_token_expired = refresh_token_expires_in
existing_oauth.material_auth_status = material_auth_status
update_redis_ids.append(existing_oauth.id)
else:
# 新增记录
oauth_id = generate_id()
update_redis_ids.append(oauth_id)
db.add(UserOAuth(
id=generate_id(),
id=oauth_id,
account_id=str(account_id),
account_name=account.get("account_name", ""),
account_role=account.get("account_role", ""),
@@ -246,8 +260,45 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int,
material_auth_status=material_auth_status,
))
await db.commit()
for oauth_id in update_redis_ids:
await _update_redis_token(oauth_id, access_token, expires_in)
#5.本次更新成功以后，判断是否有其他同一个appid，同一个授权登录账号的授权记录,如果有，则更新token信息
from sqlalchemy import update
related_oauths = await db.execute(
select(UserOAuth).where(
UserOAuth.account_username == account_username,
UserOAuth.account_userid == account_userid,
UserOAuth.appid == app_id,
UserOAuth.user_id != user_id,
UserOAuth.deleted_at.is_(None),
)
)
related_oauths = related_oauths.scalars().all()
if related_oauths:
await db.execute(
update(UserOAuth).where(
UserOAuth.account_username == account_username,
UserOAuth.account_userid == account_userid,
UserOAuth.appid == app_id,
UserOAuth.user_id != user_id,
UserOAuth.deleted_at.is_(None),
).values(
access_token=access_token,
access_token_expired=expires_in,
refresh_token=refresh_token,
refresh_token_expired=refresh_token_expires_in,
material_auth_status=material_auth_status,
)
)
await db.commit()
for related_oauth in related_oauths:
await _update_redis_token(related_oauth.id, access_token, expires_in)
#5.返回成功
return {"message": "授权成功"}
async def get_kuaishou_token(app_id: str, secret: str, code: str, oauth_type: int, user_id: str, db: AsyncSession) -> dict:
video-gen-api/app/services/video_gen.py
+5 -2
View File
@@ -126,12 +126,15 @@ async def submit_video_task(
for ref in refs:
ref_type = ref.get("type")
ref_url = ref.get("url", "")
ref_role = ref.get("role")
if ref_type == "image" and ref_url:
resolved = _resolve_url(ref_url)
content.append({"type": "image_url", "image_url": {"url": resolved},"role":"reference_image"})
role = ref_role if ref_role in ("first_frame", "last_frame") else "reference_image"
content.append({"type": "image_url", "image_url": {"url": resolved}, "role": role})
elif ref_type == "video" and ref_url:
resolved = _resolve_url(ref_url)
content.append({"type": "video_url", "video_url": {"url": resolved},"role":"reference_video"})
role = ref_role if ref_role else "reference_video"
content.append({"type": "video_url", "video_url": {"url": resolved}, "role": role})
except (json.JSONDecodeError, TypeError):
pass
video-gen-api/app/tasks/celery_app.py
+39 -21
View File
@@ -4,6 +4,7 @@ from celery import Celery
from celery.signals import worker_process_init, worker_process_shutdown, worker_ready
from app.config import settings
from app.enums.celery_queue import CeleryQueue, CeleryTaskName
from app.models.base import engine
from app.tasks.async_runner import close_loop, run_async
@@ -26,7 +27,7 @@ CELERY_TASK_IMPORTS = (
)
RECOVERY_QUEUE = settings.CELERY_RECOVERY_QUEUE or "gen_recovery"
RECOVERY_QUEUE = settings.CELERY_RECOVERY_QUEUE or CeleryQueue.GEN_RECOVERY.value
def _derive_redis_db(url: str, db_no: int) -> str:
@@ -39,6 +40,21 @@ def _derive_redis_db(url: str, db_no: int) -> str:
return url.rstrip("/") + f"/{db_no}"
def _beat_schedule() -> dict:
if not bool(getattr(settings, "POLL_DUE_DISPATCH_ENABLED", True)):
return {}
return {
"dispatch-due-poll-tasks-every-minute": {
"task": CeleryTaskName.DISPATCH_DUE_POLL.value,
"schedule": max(1, int(settings.POLL_DUE_DISPATCH_INTERVAL_SECONDS or 60)),
"options": {
"queue": RECOVERY_QUEUE,
"priority": settings.DOWNLOAD_TASK_PRIORITY_RECOVER,
},
}
}
broker_url = settings.CELERY_BROKER_URL or (_derive_redis_db(settings.REDIS_URL, 1) if settings.REDIS_URL else "")
backend_url = settings.CELERY_RESULT_BACKEND or (_derive_redis_db(settings.REDIS_URL, 2) if settings.REDIS_URL else "")
@@ -58,6 +74,7 @@ if broker_url:
task_acks_late=True,
task_reject_on_worker_lost=True,
task_track_started=True,
beat_schedule=_beat_schedule(),
task_annotations={
# 生成链路任务以数据库状态为准，不依赖 Celery result backend。
# 这里忽略结果可避免任务误返回 ORM / 非 JSON 对象时触发结果序列化失败。
@@ -80,24 +97,25 @@ if broker_url:
"sep": ":",
},
task_routes={
"generation.chatapi_create_generation_task": {"queue": "gen_chatapi_create"},
"generation.poll_generation_task": {"queue": "gen_provider_poll"},
"generation.download_generation_result_task": {"queue": "gen_result_download"},
"hot_opening.start_image_prompt_optimize": {"queue": "gen_chatapi_create"},
"hot_opening.start_video_prompt_optimize": {"queue": "gen_chatapi_create"},
"shot_replicate.analyze_original_video": {"queue": "gen_chatapi_create"},
"shot_replicate.analyze_custom_segment_video": {"queue": "gen_chatapi_create"},
"shot_replicate.split_one_segment": {"queue": "gen_result_download"},
"shot_replicate.start_image_prompt_optimize": {"queue": "gen_chatapi_create"},
"shot_replicate.start_video_prompt_optimize": {"queue": "gen_chatapi_create"},
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.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},
"shot_replicate.analyze_original_video": {"queue": CeleryQueue.GEN_CHATAPI_CREATE.value},
"shot_replicate.analyze_custom_segment_video": {"queue": CeleryQueue.GEN_CHATAPI_CREATE.value},
"shot_replicate.split_one_segment": {"queue": CeleryQueue.GEN_RESULT_DOWNLOAD.value},
"shot_replicate.start_image_prompt_optimize": {"queue": CeleryQueue.GEN_CHATAPI_CREATE.value},
"shot_replicate.start_video_prompt_optimize": {"queue": CeleryQueue.GEN_CHATAPI_CREATE.value},
# 恢复扫描统一走独立队列，避免占用下载/轮询/创建业务 worker。
"recovery.startup_recovery_once": {"queue": RECOVERY_QUEUE},
"shot_replicate.recover_split_tasks_once": {"queue": RECOVERY_QUEUE},
"generation.recover_download_tasks_once": {"queue": RECOVERY_QUEUE},
"generation.recover_generation_tasks_once": {"queue": RECOVERY_QUEUE},
"module_async.recover_module_async_tasks_once": {"queue": RECOVERY_QUEUE},
"user_oauth.update_oauth_accounts": {"queue": "default"},
"app.tasks.cleanup.*": {"queue": "default"},
CeleryTaskName.STARTUP_RECOVERY.value: {"queue": RECOVERY_QUEUE},
CeleryTaskName.SHOT_SPLIT_RECOVERY.value: {"queue": RECOVERY_QUEUE},
CeleryTaskName.RECOVER_DOWNLOAD.value: {"queue": RECOVERY_QUEUE},
CeleryTaskName.RECOVER_GENERATION.value: {"queue": RECOVERY_QUEUE},
CeleryTaskName.MODULE_ASYNC_RECOVERY.value: {"queue": RECOVERY_QUEUE},
"user_oauth.update_oauth_accounts": {"queue": CeleryQueue.DEFAULT.value},
"app.tasks.cleanup.*": {"queue": CeleryQueue.DEFAULT.value},
},
)
else:
@@ -121,8 +139,8 @@ def on_worker_ready(sender=None, **kwargs):
"""Celery worker 启动时做一次容灾恢复。
注意：
- 不启用 Celery beat。
- 启动容灾保留，但只投递一个 recovery.startup_recovery_once 协调任务。
- 启动容灾只投递一个 recovery.startup_recovery_once 协调任务。
- Celery Beat 只用于每分钟触发轻量 generation.dispatch_due_poll_tasks，不跑完整启动容灾。
- 协调任务走独立 gen_recovery 队列，串行扫描并把真实业务任务投回原队列。
- 所有 worker 都尝试抢 Redis 投递锁，只有抢到锁的 worker 投递恢复任务。
"""
@@ -182,4 +200,4 @@ def on_worker_process_shutdown(**kwargs):
except Exception:
pass
finally:
close_loop()
close_loop()
video-gen-api/app/tasks/generation_create_tasks.py
+63 -35
View File
@@ -6,21 +6,31 @@ from typing import Any, Optional
from sqlalchemy import select
from app.config import settings
from app.enums.celery_queue import CeleryQueue
from app.enums.generation_task import (
ALLOWED_GENERATION_MODES,
ChatGenerationPipelineStage,
ChatGenerationTaskEventType,
ChatGenerationTaskStatus,
GenerationMode,
GenerationType,
)
from app.models.base import async_session
from app.models.chat_generation_task import ChatGenerationTask
from app.services.error_codes import extract_error_message
from app.services.generation_log_service import log_task_event
from app.services.generation_poll_schedule_service import ensure_video_poll_fields
from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once
from app.services.generation_provider_service import create_provider_task
from app.services.media_token_usage_snapshot_service import sync_chat_generation_task_media_token_snapshot
from app.services.redis_registry_service import ensure_aware_utc
from app.tasks.celery_app import celery_app
ALLOWED_GENERATION_MODES = {"chatapi_async", "hot_opening_replicate", "shot_replicate"}
def _now() -> datetime:
return datetime.now(timezone.utc)
def _get_first_value(obj: Any, *field_names: str) -> Optional[Any]:
"""
兼容不同版本字段名，避免字段调整后 Celery 任务直接报错。
@@ -74,7 +84,7 @@ def _build_optimized_prompt_by_params(task: ChatGenerationTask) -> str:
# 爆款开头复刻第5步的视频生成，original_prompt 已经是视频提词 JSON schema。
# 不能再追加“时长/比例/分辨率”中文参数，否则会污染 schema。
if generation_mode in {"hot_opening_replicate", "shot_replicate"} and gen_type == "video":
if generation_mode in {GenerationMode.HOT_OPENING_REPLICATE.value, GenerationMode.SHOT_REPLICATE.value} and gen_type == GenerationType.VIDEO.value:
stripped = base_prompt.strip()
if stripped.startswith("{") or stripped.startswith("["):
return base_prompt
@@ -88,25 +98,25 @@ def _build_optimized_prompt_by_params(task: ChatGenerationTask) -> str:
parts = []
if gen_type == "video":
if gen_type == GenerationType.VIDEO.value:
# 时长：4秒，画面比例：16:9，分辨率：480p
if duration:
parts.append(f"时长：{duration}秒")
parts.append(f"画面比例：{aspect_ratio}")
parts.append(f"分辨率：{resolution}")
else:
parts.append(f"时长：4秒")
parts.append(f"画面比例：16:9")
parts.append(f"分辨率：480p")
elif gen_type == "image":
if image_size :
parts.append("时长：4秒")
parts.append("画面比例：16:9")
parts.append("分辨率：480p")
elif gen_type == GenerationType.IMAGE.value:
if image_size:
parts.append(f"分辨率：{image_size}")
parts.append(f"画布比例：{image_proportion}")
parts.append(f"像素尺寸：{image_px}")
else:
parts.append(f"分辨率：2K")
parts.append(f"画布比例：1:1")
parts.append(f"像素尺寸：2048x2048")
parts.append("分辨率：2K")
parts.append("画布比例：1:1")
parts.append("像素尺寸：2048x2048")
else:
# 未知类型时返回原始字符
return base_prompt
@@ -131,7 +141,7 @@ async def _run(task_id: str):
if not task or task.generation_mode not in ALLOWED_GENERATION_MODES:
return
if task.status != "generating":
if task.status != ChatGenerationTaskStatus.GENERATING.value:
return
deadline_at = ensure_aware_utc(task.deadline_at)
@@ -140,28 +150,37 @@ async def _run(task_id: str):
db,
task=task,
error_message="任务超时",
pipeline_stage="timeout",
pipeline_stage=ChatGenerationPipelineStage.TIMEOUT.value,
)
await db.commit()
await log_task_event(task, event_type="TASK_TIMEOUT", to_status="failed", to_stage="timeout")
await log_task_event(
task,
event_type=ChatGenerationTaskEventType.TASK_TIMEOUT.value,
to_status=ChatGenerationTaskStatus.FAILED.value,
to_stage=ChatGenerationPipelineStage.TIMEOUT.value,
)
from app.services.generation_module_hook_service import notify_chat_generation_task_finished
await notify_chat_generation_task_finished(db, task)
await db.commit()
return
if task.pipeline_stage not in ("queued", "preparing", "creating_provider_task"):
if task.pipeline_stage not in (
ChatGenerationPipelineStage.QUEUED.value,
ChatGenerationPipelineStage.PREPARING.value,
ChatGenerationPipelineStage.CREATING_PROVIDER_TASK.value,
):
return
try:
if not task.optimized_prompt:
old_stage = task.pipeline_stage
task.pipeline_stage = "preparing"
task.pipeline_stage = ChatGenerationPipelineStage.PREPARING.value
await db.commit()
await log_task_event(
task,
event_type="PROMPT_CONCAT_START",
event_type=ChatGenerationTaskEventType.PROMPT_CONCAT_START.value,
from_stage=old_stage,
to_stage="preparing",
to_stage=ChatGenerationPipelineStage.PREPARING.value,
message="开始本地拼接提示词，不调用提词优化API",
)
@@ -174,8 +193,8 @@ async def _run(task_id: str):
await db.commit()
await log_task_event(
task,
event_type="PROMPT_CONCAT_SUCCESS",
to_stage="preparing",
event_type=ChatGenerationTaskEventType.PROMPT_CONCAT_SUCCESS.value,
to_stage=ChatGenerationPipelineStage.PREPARING.value,
detail={
"optimized_prompt": optimized_prompt,
"gen_type": task.gen_type,
@@ -184,11 +203,14 @@ async def _run(task_id: str):
)
if task.seedance_task_id or task.provider_task_id:
task.pipeline_stage = "waiting_remote"
task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value
if task.gen_type == GenerationType.VIDEO.value:
ensure_video_poll_fields(task, now=_now())
task.next_poll_at = _now()
await db.commit()
elif task.remote_result_url:
task.pipeline_stage = "result_ready"
task.pipeline_stage = ChatGenerationPipelineStage.RESULT_READY.value
await db.commit()
from app.tasks.generation_download_tasks import enqueue_download_task
@@ -198,13 +220,13 @@ async def _run(task_id: str):
else:
old_stage = task.pipeline_stage
task.pipeline_stage = "creating_provider_task"
task.pipeline_stage = ChatGenerationPipelineStage.CREATING_PROVIDER_TASK.value
await db.commit()
await log_task_event(
task,
event_type="PROVIDER_CREATE_START",
event_type=ChatGenerationTaskEventType.PROVIDER_CREATE_START.value,
from_stage=old_stage,
to_stage="creating_provider_task",
to_stage=ChatGenerationPipelineStage.CREATING_PROVIDER_TASK.value,
)
created = await create_provider_task(db, task)
@@ -216,7 +238,7 @@ async def _run(task_id: str):
task.remote_result_url = created.get("remote_result_url") or task.remote_result_url
if task.gen_type == "image":
if task.gen_type == GenerationType.IMAGE.value:
task.image_tokens_used = created.get("image_tokens", task.image_tokens_used or 0) or 0
task.provider_response_json = json.dumps(
@@ -228,21 +250,26 @@ async def _run(task_id: str):
if task.remote_result_url and not task.seedance_task_id:
# 同步图片路径：原 SDK 已经返回最终 URL。
task.pipeline_stage = "result_ready"
task.pipeline_stage = ChatGenerationPipelineStage.RESULT_READY.value
else:
# 视频路径：provider 返回 task id，后续轮询。
task.pipeline_stage = "waiting_remote"
task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value
if task.gen_type == GenerationType.VIDEO.value:
current_time = _now()
ensure_video_poll_fields(task, now=current_time)
task.next_poll_at = current_time
task.poll_interval_seconds = int(task.poll_interval_seconds or 0)
task.status = "generating"
task.status = ChatGenerationTaskStatus.GENERATING.value
await db.commit()
await log_task_event(
task,
event_type="PROVIDER_CREATE_SUCCESS",
event_type=ChatGenerationTaskEventType.PROVIDER_CREATE_SUCCESS.value,
to_stage=task.pipeline_stage,
detail=created,
)
if task.pipeline_stage == "result_ready":
if task.pipeline_stage == ChatGenerationPipelineStage.RESULT_READY.value:
from app.tasks.generation_download_tasks import enqueue_download_task
await enqueue_download_task(db, task, reason="create_result_ready")
@@ -253,10 +280,11 @@ async def _run(task_id: str):
task,
reason="create_provider_success",
check_at=_now() + timedelta(seconds=int(settings.POLL_TASK_LEASE_SECONDS or 300)),
next_poll_at=getattr(task, "next_poll_at", None),
)
poll_generation_task.apply_async(
args=[task.id],
queue="gen_provider_poll",
queue=CeleryQueue.GEN_PROVIDER_POLL.value,
countdown=0,
)
@@ -278,10 +306,10 @@ async def _run(task_id: str):
db,
task=task,
error_message=error_message,
pipeline_stage="failed",
pipeline_stage=ChatGenerationPipelineStage.FAILED.value,
)
await db.commit()
await log_task_event(task, event_type="TASK_FAILED", message=task.error_message)
await log_task_event(task, event_type=ChatGenerationTaskEventType.TASK_FAILED.value, message=task.error_message)
from app.services.generation_module_hook_service import notify_chat_generation_task_finished
await notify_chat_generation_task_finished(db, task)
await db.commit()
@@ -305,4 +333,4 @@ else:
def apply_async(self, *args, **kwargs):
raise RuntimeError("Celery is disabled")
chatapi_create_generation_task = _DisabledTask()
chatapi_create_generation_task = _DisabledTask()
video-gen-api/app/tasks/generation_poll_tasks.py
+156 -41
View File
@@ -6,10 +6,28 @@ from typing import Any
from sqlalchemy import select
from app.config import settings
from app.enums.celery_queue import CeleryQueue
from app.enums.generation_task import (
ALLOWED_GENERATION_MODES,
ChatGenerationPipelineStage,
ChatGenerationTaskEventType,
ChatGenerationTaskStatus,
GenerationType,
PROVIDER_FAILED_STATUSES,
PROVIDER_SUCCESS_STATUSES,
)
from app.models.base import async_session
from app.models.chat_generation_task import ChatGenerationTask
from app.services.error_codes import extract_error_message
from app.services.generation_log_service import log_task_event, log_provider_call
from app.services.generation_poll_schedule_service import (
build_default_poll_schedule,
build_video_pending_poll_schedule,
ensure_video_poll_fields,
is_final_poll_due,
is_poll_not_due,
is_video_generation_task,
)
from app.services.generation_refund_service import mark_chat_generation_task_failed_and_refund_once
from app.services.generation_provider_service import poll_provider_task
from app.services.media_token_usage_snapshot_service import sync_chat_generation_task_media_token_snapshot
@@ -22,20 +40,19 @@ from app.services.redis_registry_service import (
)
from app.tasks.celery_app import celery_app
ALLOWED_GENERATION_MODES = {"chatapi_async", "hot_opening_replicate", "shot_replicate"}
POLL_QUEUE = "gen_provider_poll"
POLL_QUEUE = CeleryQueue.GEN_PROVIDER_POLL.value
def _now() -> datetime:
return datetime.now(timezone.utc)
def _is_success(status: str) -> bool:
return status in ("succeeded", "success", "completed", "done")
def _is_success(status: str | None) -> bool:
return str(status or "").lower() in PROVIDER_SUCCESS_STATUSES
def _is_failed(status: str) -> bool:
return status in ("failed", "error", "canceled", "cancelled")
def _is_failed(status: str | None) -> bool:
return str(status or "").lower() in PROVIDER_FAILED_STATUSES
def _engine_snapshot(task: ChatGenerationTask) -> dict:
@@ -46,8 +63,7 @@ def _engine_snapshot(task: ChatGenerationTask) -> dict:
def _deadline_expired(task: ChatGenerationTask, now: datetime | None = None) -> bool:
deadline_at = ensure_aware_utc(task.deadline_at)
return bool(deadline_at and deadline_at <= (now or _now()))
return is_final_poll_due(task, now=now)
def _poll_check_at(*, delay_seconds: int | float | None = None, now: datetime | None = None) -> datetime:
@@ -83,6 +99,8 @@ def _build_poll_active_payload(
"queue": POLL_QUEUE,
"poll_count": int(task.poll_count or 0),
"retry_count": int(task.retry_count or 0),
"poll_started_at": datetime_to_epoch(task.poll_started_at) if getattr(task, "poll_started_at", None) else None,
"poll_interval_seconds": int(getattr(task, "poll_interval_seconds", 0) or 0),
"last_poll_at": datetime_to_epoch(task.last_poll_at) if task.last_poll_at else None,
"next_poll_at": datetime_to_epoch(checked_next_poll_at) if checked_next_poll_at else None,
"deadline_at": datetime_to_epoch(task.deadline_at) if task.deadline_at else None,
@@ -154,12 +172,18 @@ async def _mark_timeout(db, task: ChatGenerationTask, *, message: str = "任务
db,
task=task,
error_message=message,
pipeline_stage="timeout",
pipeline_stage=ChatGenerationPipelineStage.TIMEOUT.value,
)
task.next_poll_at = None
await _notify_finished(db, task)
await db.commit()
await remove_poll_active(task.id)
await log_task_event(task, event_type="TASK_TIMEOUT", to_status="failed", to_stage="timeout")
await log_task_event(
task,
event_type=ChatGenerationTaskEventType.TASK_TIMEOUT.value,
to_status=ChatGenerationTaskStatus.FAILED.value,
to_stage=ChatGenerationPipelineStage.TIMEOUT.value,
)
async def _mark_failed(db, task: ChatGenerationTask, *, message: str, detail: Any = None) -> None:
@@ -167,12 +191,84 @@ async def _mark_failed(db, task: ChatGenerationTask, *, message: str, detail: An
db,
task=task,
error_message=message,
pipeline_stage="failed",
pipeline_stage=ChatGenerationPipelineStage.FAILED.value,
)
task.next_poll_at = None
await _notify_finished(db, task)
await db.commit()
await remove_poll_active(task.id)
await log_task_event(task, event_type="POLL_FAILED", message=task.error_message, detail=detail)
await log_task_event(
task,
event_type=ChatGenerationTaskEventType.POLL_FAILED.value,
message=task.error_message,
detail=detail,
)
async def _skip_not_due(task: ChatGenerationTask) -> None:
next_poll_at = ensure_aware_utc(task.next_poll_at)
if next_poll_at is None:
return
await register_poll_active(
task,
check_at=next_poll_at,
next_poll_at=next_poll_at,
reason="poll_task_not_due",
)
await log_task_event(
task,
event_type=ChatGenerationTaskEventType.POLL_SKIP_NOT_DUE.value,
message="视频任务尚未到下一次轮询时间，本次 poll 跳过",
detail={"next_poll_at": next_poll_at, "pipeline_stage": task.pipeline_stage},
)
async def _schedule_next_poll(
task: ChatGenerationTask,
*,
reason: str,
default_delay_seconds: int | None = None,
) -> None:
current_time = _now()
if is_video_generation_task(task):
schedule = build_video_pending_poll_schedule(task, now=current_time)
else:
schedule = build_default_poll_schedule(
task,
now=current_time,
delay_seconds=default_delay_seconds,
reason=reason,
)
task.next_poll_at = schedule.next_poll_at
task.poll_interval_seconds = schedule.poll_interval_seconds
await log_task_event(
task,
event_type=ChatGenerationTaskEventType.POLL_SCHEDULED.value,
message=f"已登记下一次轮询。reason={schedule.reason}",
detail={
"delay_seconds": schedule.delay_seconds,
"next_poll_at": schedule.next_poll_at,
"direct_countdown": schedule.direct_countdown,
"poll_interval_seconds": schedule.poll_interval_seconds,
"source_reason": reason,
},
)
await register_poll_active(
task,
check_at=schedule.next_poll_at,
next_poll_at=schedule.next_poll_at,
reason=schedule.reason,
)
if schedule.direct_countdown:
poll_generation_task.apply_async(
args=[task.id],
queue=POLL_QUEUE,
countdown=max(0, int(schedule.delay_seconds)),
)
async def _run(task_id: str):
@@ -190,11 +286,23 @@ async def _run(task_id: str):
return
# 只处理正在生成，且处于远程等待/轮询中的任务。
if task.status != "generating" or task.pipeline_stage not in ("waiting_remote", "polling"):
if task.status != ChatGenerationTaskStatus.GENERATING.value or task.pipeline_stage not in (
ChatGenerationPipelineStage.WAITING_REMOTE.value,
ChatGenerationPipelineStage.POLLING.value,
):
await remove_poll_active(task.id)
return
final_poll_before_timeout = _deadline_expired(task)
current_time = _now()
if is_video_generation_task(task):
ensure_video_poll_fields(task, now=current_time)
if is_poll_not_due(task, now=current_time):
await db.commit()
await _skip_not_due(task)
return
await db.commit()
final_poll_before_timeout = _deadline_expired(task, current_time)
if final_poll_before_timeout and not (task.seedance_task_id or task.provider_task_id):
await _mark_timeout(db, task, message="任务轮询超时")
return
@@ -206,21 +314,23 @@ async def _run(task_id: str):
if final_poll_before_timeout:
await log_task_event(
task,
event_type="FINAL_POLL_BEFORE_TIMEOUT",
event_type=ChatGenerationTaskEventType.FINAL_POLL_BEFORE_TIMEOUT.value,
message="任务已到 deadline，执行最后一次供应商查询后再判定超时",
detail={"deadline_at": task.deadline_at, "stage": task.pipeline_stage},
)
# 标记本次正在轮询，并登记 poll lease。
# 如果 worker 在供应商接口调用过程中退出，启动恢复会在 lease 过期后重新投递。
task.pipeline_stage = "polling"
task.pipeline_stage = ChatGenerationPipelineStage.POLLING.value
task.poll_count = (task.poll_count or 0) + 1
task.last_poll_at = _now()
task.next_poll_at = _poll_lease_until(task.last_poll_at)
await db.commit()
await register_poll_active(
task,
check_at=_poll_lease_until(task.last_poll_at),
reason="polling_lease",
next_poll_at=task.next_poll_at,
)
try:
@@ -247,7 +357,7 @@ async def _run(task_id: str):
)
if _is_success(status):
if task.gen_type == "image":
if task.gen_type == GenerationType.IMAGE.value:
task.remote_result_url = poll_result.get("image_url")
task.image_tokens_used = poll_result.get("image_tokens", 0) or 0
else:
@@ -261,12 +371,13 @@ async def _run(task_id: str):
await _mark_failed(db, task, message="供应商任务成功但未返回结果URL", detail=poll_result)
return
task.pipeline_stage = "result_ready"
task.pipeline_stage = ChatGenerationPipelineStage.RESULT_READY.value
task.retry_count = 0
task.next_poll_at = None
await db.commit()
await remove_poll_active(task.id)
await log_task_event(task, event_type="POLL_SUCCESS", to_stage="result_ready")
await log_task_event(task, event_type=ChatGenerationTaskEventType.POLL_SUCCESS.value, to_stage=ChatGenerationPipelineStage.RESULT_READY.value)
from app.tasks.generation_download_tasks import enqueue_download_task
@@ -286,7 +397,7 @@ async def _run(task_id: str):
if final_poll_before_timeout:
await log_task_event(
task,
event_type="FINAL_POLL_BEFORE_TIMEOUT_PENDING",
event_type=ChatGenerationTaskEventType.FINAL_POLL_BEFORE_TIMEOUT_PENDING.value,
message=f"最终查询后供应商仍未完成，按超时处理。status={status}",
detail=poll_result,
)
@@ -294,26 +405,17 @@ async def _run(task_id: str):
return
# 供应商仍在 pending / running 时，把阶段从 polling 改回 waiting_remote。
# 同时登记下一次 poll active，Celery countdown 丢失时可由恢复任务拉起。
task.pipeline_stage = "waiting_remote"
# 视频任务写入 next_poll_at，由 Beat dispatcher 到期投递；短间隔可保留 countdown 兼容。
task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value
task.retry_count = 0
await _schedule_next_poll(task, reason="poll_pending_next")
await db.commit()
await log_task_event(task, event_type="POLL_PENDING", message=f"status={status}")
delay_seconds = int(settings.CHATAPI_ASYNC_POLL_INTERVAL_SECONDS or 30)
next_poll_at = _now() + timedelta(seconds=max(1, delay_seconds))
await register_poll_active(
await log_task_event(
task,
check_at=_poll_check_at(delay_seconds=delay_seconds),
next_poll_at=next_poll_at,
reason="poll_pending_next",
)
poll_generation_task.apply_async(
args=[task.id],
queue=POLL_QUEUE,
countdown=delay_seconds,
event_type=ChatGenerationTaskEventType.POLL_PENDING.value,
message=f"status={status}",
detail={"next_poll_at": task.next_poll_at, "poll_interval_seconds": task.poll_interval_seconds},
)
except Exception as exc:
@@ -331,7 +433,7 @@ async def _run(task_id: str):
if final_poll_before_timeout:
await log_task_event(
task,
event_type="FINAL_POLL_BEFORE_TIMEOUT_ERROR",
event_type=ChatGenerationTaskEventType.FINAL_POLL_BEFORE_TIMEOUT_ERROR.value,
message=str(exc),
)
await _mark_timeout(db, task, message="任务轮询超时")
@@ -339,21 +441,34 @@ async def _run(task_id: str):
task.retry_count = (task.retry_count or 0) + 1
# 视频轮询的临时异常不再 3 次内直接退款；继续降频到 24 小时最终 deadline。
if is_video_generation_task(task):
task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value
await _schedule_next_poll(task, reason="poll_exception_retry")
await db.commit()
return
if task.retry_count > settings.CHATAPI_ASYNC_MAX_RETRIES:
error_message = extract_error_message(exc, "轮询") if callable(extract_error_message) else str(exc)
await _mark_failed(db, task, message=error_message)
else:
# 临时轮询异常时，不让任务停在 polling。
# 回到 waiting_remote，等待下一次重试轮询。
task.pipeline_stage = "waiting_remote"
task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value
delay_seconds = int(settings.CHATAPI_ASYNC_RETRY_BACKOFF_SECONDS or 30) * int(task.retry_count or 1)
default_schedule = build_default_poll_schedule(
task,
now=_now(),
delay_seconds=delay_seconds,
reason="poll_exception_retry",
)
task.next_poll_at = default_schedule.next_poll_at
await db.commit()
delay_seconds = int(settings.CHATAPI_ASYNC_RETRY_BACKOFF_SECONDS or 30) * int(task.retry_count or 1)
next_poll_at = _now() + timedelta(seconds=max(1, delay_seconds))
await register_poll_active(
task,
check_at=_poll_check_at(delay_seconds=delay_seconds),
next_poll_at=next_poll_at,
next_poll_at=default_schedule.next_poll_at,
reason="poll_exception_retry",
)
video-gen-api/app/tasks/generation_recovery_tasks.py
+59 -2
View File
@@ -5,6 +5,7 @@ import logging
from typing import Any, Awaitable, Callable, Dict
from app.config import settings
from app.enums.celery_queue import CeleryQueue
from app.models.base import async_session
from app.services.redis_registry_service import get_registry_redis, redis_acquire_lock, redis_release_lock
from app.tasks.async_runner import run_async
@@ -12,7 +13,7 @@ from app.tasks.celery_app import celery_app
logger = logging.getLogger("video_gen")
RECOVERY_QUEUE = settings.CELERY_RECOVERY_QUEUE or "gen_recovery"
RECOVERY_QUEUE = settings.CELERY_RECOVERY_QUEUE or CeleryQueue.GEN_RECOVERY.value
RecoveryRunner = Callable[[], Awaitable[Dict[str, Any]]]
@@ -30,6 +31,13 @@ async def _run_generation_once() -> Dict[str, Any]:
return await recover_generation_tasks_once(db)
async def _run_due_poll_dispatch_once() -> Dict[str, Any]:
from app.services.generation_recovery_service import dispatch_due_poll_tasks_once
async with async_session() as db:
return await dispatch_due_poll_tasks_once(db)
async def _run_module_async_once() -> Dict[str, Any]:
from app.services.module_async_recovery_service import recover_module_async_tasks_once
@@ -49,6 +57,7 @@ async def _run_with_execution_lock(
lock_key: str,
log_context: str,
runner: RecoveryRunner,
ttl_seconds: int | None = None,
) -> Dict[str, Any]:
"""恢复任务执行锁。
@@ -61,7 +70,7 @@ async def _run_with_execution_lock(
if redis is not None:
token = await redis_acquire_lock(
lock_key=lock_key,
ttl_seconds=int(settings.CELERY_RECOVERY_TASK_LOCK_TTL_SECONDS or 600),
ttl_seconds=int(ttl_seconds or settings.CELERY_RECOVERY_TASK_LOCK_TTL_SECONDS or 600),
log_context=log_context,
)
if not token:
@@ -79,6 +88,36 @@ async def _run_with_execution_lock(
await redis_release_lock(lock_key=lock_key, token=token, log_context=log_context)
async def _is_lock_held(lock_key: str) -> bool:
redis = await get_registry_redis()
if redis is None:
return False
try:
return bool(await redis.exists(lock_key))
except Exception:
return False
async def _startup_or_generation_recovery_running() -> str | None:
# Beat 触发 dispatcher 时，如果启动容灾或完整生成容灾还在跑，直接跳过本轮。
# gen_recovery concurrency=1 已经能串行；这里是多机部署、残留消息、手动触发时的双保险。
lock_checks = [
("startup_recovery", settings.CELERY_RECOVERY_STARTUP_TASK_LOCK_KEY),
("generation_recovery", settings.GENERATION_RECOVERY_LOCK_KEY),
]
for name, lock_key in lock_checks:
if await _is_lock_held(lock_key):
return name
return None
async def _run_due_poll_dispatch_with_guard() -> Dict[str, Any]:
running = await _startup_or_generation_recovery_running()
if running:
return {"skipped": "recovery_lock_held", "lock": running}
return await _run_due_poll_dispatch_once()
async def _acquire_download_recovery_loop_lock() -> tuple[bool, str]:
"""下载恢复循环锁。
@@ -220,6 +259,23 @@ if celery_app:
)
)
@celery_app.task(
name="generation.dispatch_due_poll_tasks",
bind=True,
soft_time_limit=settings.CELERY_RECOVERY_SOFT_TIME_LIMIT_SECONDS,
time_limit=settings.CELERY_RECOVERY_TIME_LIMIT_SECONDS,
)
def dispatch_due_poll_tasks(self) -> Dict[str, Any]:
return run_async(
_run_with_execution_lock(
lock_key=settings.POLL_DUE_DISPATCH_LOCK_KEY,
log_context="due_poll_dispatch",
runner=_run_due_poll_dispatch_with_guard,
ttl_seconds=int(settings.POLL_DUE_DISPATCH_LOCK_TTL_SECONDS or 55),
)
)
else:
class _DisabledTask:
@@ -232,3 +288,4 @@ else:
startup_recovery_once = _DisabledTask()
recover_download_tasks_once = _DisabledTask()
recover_generation_tasks_once = _DisabledTask()
dispatch_due_poll_tasks = _DisabledTask()
video-gen-api/app/tasks/token_refresh_task.py
+95 -34
View File
@@ -60,12 +60,33 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe
#如果code=40103或者40107，传入refresh_token已失效，失效原因一般是由于refresh_token已被使用，或授权账号重新授权并生成了新的Token
if data.get("code") in [40103, 40107]:
#清空数据库中的token信息，和Redis缓存中的token
oauth.access_token = None
oauth.access_token_expired = None
oauth.refresh_token = None
oauth.refresh_token_expired = None
from sqlalchemy import update
where_cond = UserOAuth.deleted_at.is_(None)
if oauth.appid:
where_cond = where_cond & (UserOAuth.appid == oauth.appid)
if oauth.account_username:
where_cond = where_cond & (UserOAuth.account_username == oauth.account_username)
if oauth.account_userid:
where_cond = where_cond & (UserOAuth.account_userid == oauth.account_userid)
await db.execute(
update(UserOAuth).where(where_cond).values(
access_token=None,
access_token_expired=None,
refresh_token=None,
refresh_token_expired=None,
)
)
await db.commit()
await _update_redis_token(oauth.id, "", None)
related_oauth_ids = await db.execute(
select(UserOAuth.id).where(where_cond)
)
related_oauth_ids = [row[0] for row in related_oauth_ids.all()]
for related_id in related_oauth_ids:
await _update_redis_token(related_id, "", None)
return
@@ -75,15 +96,34 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe
expires_in = datetime.now(tz=oauth.access_token_expired.tzinfo) + timedelta(seconds=data.get("expires_in", 0))
refresh_token_expires_in = datetime.now(tz=oauth.refresh_token_expired.tzinfo) + timedelta(seconds=data.get("refresh_token_expires_in", 0))
oauth.access_token = new_access_token
oauth.access_token_expired = expires_in
oauth.refresh_token = new_refresh_token
oauth.refresh_token_expires_in = refresh_token_expires_in
from sqlalchemy import update
where_cond = UserOAuth.deleted_at.is_(None)
if oauth.appid:
where_cond = where_cond & (UserOAuth.appid == oauth.appid)
if oauth.account_username:
where_cond = where_cond & (UserOAuth.account_username == oauth.account_username)
if oauth.account_userid:
where_cond = where_cond & (UserOAuth.account_userid == oauth.account_userid)
await db.execute(
update(UserOAuth).where(where_cond).values(
access_token=new_access_token,
access_token_expired=expires_in,
refresh_token=new_refresh_token,
refresh_token_expired=refresh_token_expires_in,
)
)
await db.commit()
await _update_redis_token(oauth.id, new_access_token, expires_in)
related_oauth_ids = await db.execute(
select(UserOAuth.id).where(where_cond)
)
related_oauth_ids = [row[0] for row in related_oauth_ids.all()]
for related_id in related_oauth_ids:
await _update_redis_token(related_id, new_access_token, expires_in)
logger.info(f"成功刷新巨量引擎token: oauth_id={oauth.id}, account_id={oauth.account_id}")
logger.info(f"成功刷新巨量引擎token: oauth_id={oauth.id}, account_id={oauth.account_id}, 关联账户数={len(related_oauth_ids)}")
except httpx.HTTPError as e:
logger.error(f"HTTP请求失败: oauth_id={oauth.id}, 错误: {str(e)}")
except Exception as e:
@@ -97,37 +137,39 @@ async def check_and_refresh_tokens():
query = select(UserOAuth).where(
UserOAuth.deleted_at.is_(None),
UserOAuth.access_token.is_not(None),
UserOAuth.access_token_expired.is_not(None),
UserOAuth.refresh_token.is_not(None),
UserOAuth.refresh_token_expired.is_not(None),
UserOAuth.refresh_token_expired > now,
)
result = await db.execute(query)
oauth_list = result.scalars().all()
refreshed_keys = set()
for oauth in oauth_list:
try:
#1.检查access_token是否过期，如果未过期，并且大于800秒，直接跳过不处理
if not oauth.access_token_expired:
#检查是否为支持的平台（巨量引擎）
# port_type: 平台端口（1=巨量，2=磁力，3=巨量星图，4=服务单，5=腾讯）
if oauth.port_type not in [1]:
continue
remaining_seconds = (oauth.access_token_expired - now).total_seconds()
#构建登录账号唯一标识，同一登录账号共享token
key_parts = []
if oauth.appid:
key_parts.append(oauth.appid)
if oauth.account_username:
key_parts.append(oauth.account_username)
if oauth.account_userid:
key_parts.append(oauth.account_userid)
login_key = "|".join(key_parts)
# access_token剩余时间大于等于800秒，不需要刷新
if remaining_seconds >= REFRESH_THRESHOLD_SECONDS:
#同一登录账号已刷新过，直接跳过（避免使用旧数据判断）
if login_key in refreshed_keys:
logger.debug(f"跳过重复刷新: oauth_id={oauth.id}, 同一登录账号已刷新")
continue
#2.如果access_token过期，或者剩余时间小于800秒，需要刷新token
#3.如果需要刷新token，检查refresh_token是否过期，如果refresh_token过期，说明不可刷新，需要直接重新授权，直接跳过不处理
if not oauth.refresh_token_expired:
continue
refresh_remaining_seconds = (oauth.refresh_token_expired - now).total_seconds()
if refresh_remaining_seconds <= 0:
continue
#5.获取应用配置
#获取应用配置
app_result = await db.execute(
select(UserOAuthApp).where(UserOAuthApp.app_id == oauth.appid)
)
@@ -135,12 +177,31 @@ async def check_and_refresh_tokens():
if not app:
continue
#检查是否为支持的平台（巨量引擎）
# port_type: 平台端口（1=巨量，2=磁力，3=巨量星图，4=服务单，5=腾讯）
if oauth.port_type in [1]:
#刷新token
await refresh_juliang_token(oauth, app, db)
#检查access_token是否需要刷新
need_refresh = False
# access_token为空，需要刷新
if not oauth.access_token:
need_refresh = True
# access_token_expired为空，需要刷新
elif not oauth.access_token_expired:
need_refresh = True
# access_token即将过期（剩余时间小于800秒），需要刷新
else:
remaining_seconds = (oauth.access_token_expired - now).total_seconds()
if remaining_seconds < REFRESH_THRESHOLD_SECONDS:
need_refresh = True
if not need_refresh:
continue
#refresh_token已在查询条件中过滤，确保有效才能刷新
#刷新token
await refresh_juliang_token(oauth, app, db)
refreshed_keys.add(login_key)
except Exception as e:
#7.增加错误日志
video-gen-api/app/utils/douyinRequest.py
+66 -19
View File
@@ -89,7 +89,6 @@ class DouyinRequest:
if token:
return token
async with async_session() as db:
oauth_data = await db.execute(
select(UserOAuth).where(
@@ -103,17 +102,30 @@ class DouyinRequest:
raise ValueError("无效的oauth_id")
if force_refresh:
where_cond = UserOAuth.deleted_at.is_(None)
if oauth_data.appid:
where_cond = where_cond & (UserOAuth.appid == oauth_data.appid)
if oauth_data.account_username:
where_cond = where_cond & (UserOAuth.account_username == oauth_data.account_username)
if oauth_data.account_userid:
where_cond = where_cond & (UserOAuth.account_userid == oauth_data.account_userid)
await db.execute(
update(UserOAuth).where(UserOAuth.id == oauth_id).values(
update(UserOAuth).where(where_cond).values(
access_token=None,
access_token_expired=None,
)
)
await db.commit()
await self._delete_redis_token(oauth_id)
related_oauth_ids = await db.execute(
select(UserOAuth.id).where(where_cond)
)
related_oauth_ids = [row[0] for row in related_oauth_ids.all()]
for related_id in related_oauth_ids:
await self._delete_redis_token(related_id)
new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token)
await self._set_redis_token(oauth_id, new_token, new_expired_at)
return new_token
if oauth_data.access_token_expired and oauth_data.access_token_expired > datetime.now(timezone.utc):
@@ -122,14 +134,46 @@ class DouyinRequest:
await self._set_redis_token(oauth_id, token, expired_at)
return token
where_cond = UserOAuth.deleted_at.is_(None)
if oauth_data.appid:
where_cond = where_cond & (UserOAuth.appid == oauth_data.appid)
if oauth_data.account_username:
where_cond = where_cond & (UserOAuth.account_username == oauth_data.account_username)
if oauth_data.account_userid:
where_cond = where_cond & (UserOAuth.account_userid == oauth_data.account_userid)
related_oauths = await db.execute(
select(UserOAuth.access_token, UserOAuth.access_token_expired).where(
where_cond,
UserOAuth.access_token_expired.is_not(None),
UserOAuth.access_token_expired > datetime.now(timezone.utc),
).limit(1)
)
related_oauth = related_oauths.first()
if related_oauth:
token, expired_at = related_oauth
await self._set_redis_token(oauth_id, token, expired_at)
return token
if oauth_data.refresh_token_expired and oauth_data.refresh_token_expired < datetime.now(timezone.utc):
raise ValueError("授权已过期，请重新授权")
new_token, new_expired_at = await self.refresh_access_token(db, oauth_id, oauth_data.appid, oauth_data.refresh_token)
await self._set_redis_token(oauth_id, new_token, new_expired_at)
return new_token
async def refresh_access_token(self, db: AsyncSession, oauth_id: str, appid: str, refresh_token: str) -> Tuple[str, datetime]:
oauth_info = await db.execute(
select(UserOAuth.account_username, UserOAuth.account_userid).where(
UserOAuth.id == oauth_id,
UserOAuth.deleted_at.is_(None),
).limit(1)
)
oauth_info = oauth_info.first()
if not oauth_info:
raise ValueError("无效的oauth_id")
account_username, account_userid = oauth_info
result = await db.execute(
select(UserOAuthApp.secret).where(
UserOAuthApp.app_id == appid,
@@ -169,9 +213,17 @@ class DouyinRequest:
new_expired_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in)
where_cond = UserOAuth.deleted_at.is_(None)
if appid:
where_cond = where_cond & (UserOAuth.appid == appid)
if account_username:
where_cond = where_cond & (UserOAuth.account_username == account_username)
if account_userid:
where_cond = where_cond & (UserOAuth.account_userid == account_userid)
await db.execute(
update(UserOAuth).where(
UserOAuth.id == oauth_id,
where_cond,
).values(
access_token=new_access_token,
refresh_token=new_refresh_token,
@@ -181,6 +233,14 @@ class DouyinRequest:
)
await db.commit()
related_oauth_ids = await db.execute(
select(UserOAuth.id).where(where_cond)
)
related_oauth_ids = [row[0] for row in related_oauth_ids.all()]
for related_id in related_oauth_ids:
await self._set_redis_token(related_id, new_access_token, new_expired_at)
return new_access_token, new_expired_at
# 有token请求
@@ -267,19 +327,6 @@ class DouyinRequest:
else:
raise ValueError('网络错误，稍后重试。')
# options_log = {}
# if options:
# for key, value in options.items():
# if key == 'files':
# options_log[key] = {k: (v[0], 'bytes_content', v[2]) for k, v in value.items()}
# else:
# options_log[key] = value
# raise RuntimeError(
# f'DouYin API request failed after 5 retries. '
# f'url:{url};oauthId:{oauth_id};options:{json.dumps(options_log)};response:{res}'
# )
# if code != 0:
# raise ValueError(f'response:{res}')
# 无token请求
async def request_with_context(
video-gen-app/dist/assets/index-B5pqnZWD.js
Vendored
-442
View File
File diff suppressed because one or more lines are too long
video-gen-app/dist/assets/index-BhPFzWLH.css
Vendored
-1
View File
File diff suppressed because one or more lines are too long
video-gen-app/dist/assets/index-Cr8hE_b3.js
Vendored
+442
View File
File diff suppressed because one or more lines are too long
video-gen-app/dist/assets/index-DjAbNPLm.js
Vendored
-442
View File
File diff suppressed because one or more lines are too long
video-gen-app/dist/assets/index-lZSCkINs.js
Vendored
-442
View File
File diff suppressed because one or more lines are too long
video-gen-app/dist/index.html
Vendored
+1 -1
View File
@@ -28,7 +28,7 @@
}
})();
</script>
<script type="module" crossorigin src="/assets/index-DjAbNPLm.js"></script>
<script type="module" crossorigin src="/assets/index-Cr8hE_b3.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-Bi8mcSs8.css">
</head>
<body>
video-gen-app/src/components/PreResultDisplay.tsx
+7 -11
View File
@@ -55,17 +55,14 @@ const PreResultDisplay: React.FC<PreResultDisplayProps> = ({ preResult }) => {
if (!config) return null;
const value = parsedData[key as keyof PreResultData];
const isYes = value === 'YES';
const isNo = value === 'NO';
const isUnknown = value === 'UNKNOWN';
if (isUnknown) return null;
let displayLabel, displayColor;
// if (isUnknown) {
// displayLabel = config.unknownLabel;
// displayColor = config.unknownColor;
// } else
if (isYes) {
displayLabel = config.label;
displayColor = config.yesColor;
} else {
} else if (isNo) {
displayLabel = config.noLabel;
displayColor = config.noColor;
}
@@ -102,15 +99,14 @@ const PreResultDisplay: React.FC<PreResultDisplayProps> = ({ preResult }) => {
{fieldConfig.map(config => {
const value = parsedData[config.key as keyof PreResultData];
const isYes = value === 'YES';
const isNo = value === 'NO';
const isUnknown = value === 'UNKNOWN';
if (isUnknown) return null;
let displayLabel, displayColor;
if (isUnknown) {
displayLabel = config.unknownLabel;
displayColor = config.unknownColor;
} else if (isYes) {
if (isYes) {
displayLabel = config.label;
displayColor = config.yesColor;
} else {
} else if (isNo) {
displayLabel = config.noLabel;
displayColor = config.noColor;
}
video-gen-app/src/pages/GenerateConver.tsx
+892 -307
View File
File diff suppressed because it is too large Load Diff
video-gen-app/src/pages/GeneratedRecord.tsx
+245 -162
View File
@@ -171,6 +171,27 @@ const GeneratedRecord: React.FC = () => {
const hasGeneratedResourceId = (item: any): boolean => {
return Boolean(item.generatedResourceId);
};
// 检查视频尺寸是否符合480p要求（270×480或480×270）
const checkVideoSize = (imagePx: string): boolean => {
if (!imagePx) return true;
const match = imagePx.match(/^(\d+)x(\d+)$/);
if (!match) return true;
const width = parseInt(match[1], 10);
const height = parseInt(match[2], 10);
const aspectRatio = width / height;
const isLandscape = width >= height;
if (isLandscape) {
const isValidWidth = width >= 480 && width <= 2560;
const isValidHeight = height >= 270 && height <= 1440;
const isValidRatio = aspectRatio >= 1.775 && aspectRatio <= 1.784;
return isValidWidth && isValidHeight && isValidRatio;
} else {
const isValidWidth = width >= 270 && width <= 1440;
const isValidHeight = height >= 480 && height <= 2560;
const isValidRatio = aspectRatio >= 0.555 && aspectRatio <= 0.564;
return isValidWidth && isValidHeight && isValidRatio;
}
};
// 多选相关函数
const handleToggleSelect = (itemId: string) => {
setSelectedItems(prev => {
@@ -671,6 +692,34 @@ const GeneratedRecord: React.FC = () => {
const handleDateChange = (dateString: string) => {
setSelectedDate(dateString);
};
const getVideoDimensions = (videoUrl: string): Promise<string> => {
return new Promise((resolve) => {
const video = document.createElement('video');
const baseUrl = import.meta.env.VITE_API_BASE || "http://localhost:8000";
let cleanPath = videoUrl.startsWith('/') ? videoUrl.slice(1) : videoUrl;
let cleanBase = baseUrl.endsWith('/') ? baseUrl.slice(0, -1) : baseUrl;
video.src = `${cleanBase}/${cleanPath}`;
video.crossOrigin = 'anonymous';
video.onloadedmetadata = () => {
const width = video.videoWidth;
const height = video.videoHeight;
video.remove();
resolve(`${width}x${height}`);
};
video.onerror = () => {
video.remove();
resolve('');
};
video.onabort = () => {
video.remove();
resolve('');
};
setTimeout(() => {
video.remove();
resolve('');
}, 5000);
});
};
const loadRecordList = () => {
setLoading(true);
let historySource = '';
@@ -690,8 +739,16 @@ const GeneratedRecord: React.FC = () => {
if (historySource) {
parameters = `${selectedDate}?gen_type=${filterMedia}&history_source=${historySource}&page=${Pagebreak.page}&page_size=${Pagebreak.pageSize}`;
}
gethistoryItems(parameters).then((res: any) => {
gethistoryItems(parameters).then(async (res: any) => {
const data = Array.isArray(res) ? res : (res?.items || []);
for (const item of data) {
if (item.videoUrl && item.videoUrl.trim()) {
const dimensions = await getVideoDimensions(item.videoUrl);
if (dimensions) {
item.imagePx = dimensions;
}
}
}
let recordList = [{
generatedDate: res.generatedDate,
items: data,
@@ -712,11 +769,20 @@ const GeneratedRecord: React.FC = () => {
} finally {
}
} else {
gethistory(parameters).then((res: any) => {
gethistory(parameters).then(async (res: any) => {
const data = Array.isArray(res) ? res : (res?.groups || []);
data.forEach(group => {
for (const group of data) {
group.page = 1;
});
for (const item of group.items) {
if (item.videoUrl && item.videoUrl.trim()) {
const dimensions = await getVideoDimensions(item.videoUrl);
if (dimensions) {
console.log(dimensions);
item.imagePx = dimensions;
}
}
}
}
if (Pagebreak.page === 1) {
setRecordList(data);
} else {
@@ -762,7 +828,6 @@ const GeneratedRecord: React.FC = () => {
}
try {
const res: any = await gethistoryItems(parameters);
// gethistoryItems 返回数组，直接使用
const newItems: any[] = res.items || [];
if (newItems && newItems.length > 0) {
setRecordList(prev => prev.map(group => {
@@ -924,7 +989,7 @@ const GeneratedRecord: React.FC = () => {
</Button>
<Button
onClick={() => handleDeleteSelected()}
style={{ borderRadius: 8, background: 'linear-gradient(135deg, #bf0b0a 0%, #ff8165 100%)', border: 'none', boxShadow: '0 4px 16px rgba(191, 11, 10, 0.35)', fontWeight: 600, padding: '8px 24px', color: '#fff' }}
style={{ borderRadius: 8, background: 'linear-gradient(135deg, #bf0b0a 0%, #ff8165 100%)', fontWeight: 600, padding: '8px 24px', color: '#fff' }}
>
删除 ({selectedItems.size})
</Button>
@@ -1098,170 +1163,188 @@ const GeneratedRecord: React.FC = () => {
gap: 8,
}}>
{group.items.map((item: any) => {
const displayUrl = filterMedia === 'video' && item.videoCoverUrl
? buildUrl(item.videoCoverUrl, true)
: buildUrl(filterMedia === 'video' ? item.videoUrl : item.imageUrl, true);
const hasCover = filterMedia === 'video' && item.videoCoverUrl;
const showImage = hasCover || filterMedia === 'image';
const isSelectedItem = selectedItems.has(getItemResourceId(item));
return (
<div
key={getItemResourceId(item)}
className="media-card"
style={{
width: 160,
height: 120,
position: 'relative',
borderRadius: 4,
cursor: 'pointer',
overflow: 'hidden',
boxShadow: '0 2px 8px rgba(0,0,0,0.1)',
transition: 'transform 0.2s, box-shadow 0.2s',
}}
onClick={() => isSelectionMode ? handleToggleSelect(getItemResourceId(item)) : handlePreview(item)}
onMouseEnter={(e) => {
if (!isSelectionMode) {
(e.currentTarget as HTMLElement).style.transform = 'scale(1.05)';
(e.currentTarget as HTMLElement).style.boxShadow = '0 4px 16px rgba(0,0,0,0.2)';
}
}}
onMouseLeave={(e) => {
if (!isSelectionMode) {
(e.currentTarget as HTMLElement).style.transform = 'scale(1)';
(e.currentTarget as HTMLElement).style.boxShadow = '0 2px 8px rgba(0,0,0,0.1)';
}
}}
>
{showImage ? (
<img
src={displayUrl}
alt="预览"
style={{
const displayUrl = filterMedia === 'video' && item.videoCoverUrl
? buildUrl(item.videoCoverUrl, true)
: buildUrl(filterMedia === 'video' ? item.videoUrl : item.imageUrl, true);
const hasCover = filterMedia === 'video' && item.videoCoverUrl;
const showImage = hasCover || filterMedia === 'image';
const isSelectedItem = selectedItems.has(getItemResourceId(item));
return (
<div
key={getItemResourceId(item)}
className="media-card"
style={{
width: 160,
height: 120,
position: 'relative',
borderRadius: 4,
cursor: 'pointer',
overflow: 'hidden',
boxShadow: '0 2px 8px rgba(0,0,0,0.1)',
transition: 'transform 0.2s, box-shadow 0.2s',
}}
onClick={() => isSelectionMode ? handleToggleSelect(getItemResourceId(item)) : handlePreview(item)}
onMouseEnter={(e) => {
if (!isSelectionMode) {
(e.currentTarget as HTMLElement).style.transform = 'scale(1.05)';
(e.currentTarget as HTMLElement).style.boxShadow = '0 4px 16px rgba(0,0,0,0.2)';
}
}}
onMouseLeave={(e) => {
if (!isSelectionMode) {
(e.currentTarget as HTMLElement).style.transform = 'scale(1)';
(e.currentTarget as HTMLElement).style.boxShadow = '0 2px 8px rgba(0,0,0,0.1)';
}
}}
>
{showImage ? (
<img
src={displayUrl}
alt="预览"
style={{
width: '100%',
height: '100%',
objectFit: 'cover',
}}
/>
) : (
<div style={{
width: '100%',
height: '100%',
objectFit: 'cover',
}}
/>
) : (
<div style={{
width: '100%',
height: '100%',
display: 'flex',
flexDirection: 'column',
alignItems: 'center',
justifyContent: 'center',
backgroundColor: '#1e293b',
}}>
<VideoCameraOutlined style={{ color: '#64748b', fontSize: 24 }} />
<Text style={{ fontSize: 12, color: '#94a3b8', marginTop: 4 }}>暂无封面</Text>
</div>
)}
{isSelectionMode && (
<div
style={{
position: 'absolute',
top: 8,
left: 8,
width: 20,
height: 20,
borderRadius: '50%',
backgroundColor: isSelectedItem ? '#10b981' : 'rgba(255,255,255,0.9)',
border: isSelectedItem ? '2px solid #10b981' : '2px solid #d1d5db',
display: 'flex',
flexDirection: 'column',
alignItems: 'center',
justifyContent: 'center',
cursor: 'pointer',
zIndex: 10,
}}
onClick={(e) => {
e.stopPropagation();
handleToggleSelect(getItemResourceId(item));
}}
>
{isSelectedItem && (
<svg width={12} height={12} viewBox="0 0 12 12" fill="none">
<path d="M10 3L4.5 8.5L2 6" stroke="white" strokeWidth={2} strokeLinecap="round" strokeLinejoin="round" />
</svg>
)}
</div>
)}
{!isSelectionMode && (
<>
backgroundColor: '#1e293b',
}}>
<VideoCameraOutlined style={{ color: '#64748b', fontSize: 24 }} />
<Text style={{ fontSize: 12, color: '#94a3b8', marginTop: 4 }}>暂无封面</Text>
</div>
)}
{isSelectionMode && (
<>
<div
style={{
position: 'absolute',
top: 8,
left: 8,
width: 20,
height: 20,
borderRadius: '50%',
backgroundColor: isSelectedItem ? '#10b981' : 'rgba(255,255,255,0.9)',
border: isSelectedItem ? '2px solid #10b981' : '2px solid #d1d5db',
display: 'flex',
alignItems: 'center',
justifyContent: 'center',
cursor: 'pointer',
zIndex: 10,
}}
onClick={(e) => {
e.stopPropagation();
handleToggleSelect(getItemResourceId(item));
}}
>
{isSelectedItem && (
<svg width={12} height={12} viewBox="0 0 12 12" fill="none">
<path d="M10 3L4.5 8.5L2 6" stroke="white" strokeWidth={2} strokeLinecap="round" strokeLinejoin="round" />
</svg>
)}
</div>
{isSelectedItem && filterMedia === 'video' && item.imagePx && !checkVideoSize(item.imagePx) && (
<div style={{
position: 'absolute',
top: 36,
left: 8,
right: 8,
padding: '4px 8px',
backgroundColor: 'rgba(239, 68, 68, 0.9)',
color: '#fff',
fontSize: 10,
borderRadius: 4,
zIndex: 10,
}}>
视频实际尺寸可能不符合推送要求
</div>
)}
</>
)}
{!isSelectionMode && (
<>
<div style={{
position: 'absolute',
bottom: 0,
left: 0,
right: 0,
background: 'linear-gradient(transparent, rgba(0,0,0,0.5))',
padding: '8px',
color: '#fff',
fontSize: 12,
opacity: 0,
transition: 'opacity 0.2s',
}}
onMouseEnter={(e) => {
(e.currentTarget as HTMLElement).style.opacity = '1';
}}
onMouseLeave={(e) => {
(e.currentTarget as HTMLElement).style.opacity = '0';
}}
>
点击预览
</div>
<div
className="media-download-btn"
>
<Tooltip title="下载">
<div
style={{
width: 28,
height: 28,
borderRadius: '50%',
background: 'rgba(0,0,0,0.5)',
backdropFilter: 'blur(4px)',
display: 'flex',
alignItems: 'center',
justifyContent: 'center',
cursor: 'pointer',
transition: 'background 0.15s',
}}
onClick={(e) => {
e.stopPropagation();
const url = `${import.meta.env.VITE_API_BASE || "http://localhost:8000"}${item.videoUrl || item.imageUrl}&download=1`;
const link = document.createElement('a');
link.href = url;
link.download = '';
document.body.appendChild(link);
link.click();
document.body.removeChild(link);
}}
>
<DownloadOutlined style={{ color: '#fff', fontSize: 14 }} />
</div>
</Tooltip>
</div>
</>
)}
{isSelectionMode && isSelectedItem && (
<div style={{
position: 'absolute',
bottom: 0,
top: 0,
left: 0,
right: 0,
background: 'linear-gradient(transparent, rgba(0,0,0,0.5))',
padding: '8px',
color: '#fff',
fontSize: 12,
opacity: 0,
transition: 'opacity 0.2s',
}}
onMouseEnter={(e) => {
(e.currentTarget as HTMLElement).style.opacity = '1';
}}
onMouseLeave={(e) => {
(e.currentTarget as HTMLElement).style.opacity = '0';
}}
>
点击预览
</div>
<div
className="media-download-btn"
>
<Tooltip title="下载">
<div
style={{
width: 28,
height: 28,
borderRadius: '50%',
background: 'rgba(0,0,0,0.5)',
backdropFilter: 'blur(4px)',
display: 'flex',
alignItems: 'center',
justifyContent: 'center',
cursor: 'pointer',
transition: 'background 0.15s',
}}
onClick={(e) => {
e.stopPropagation();
const url = `${import.meta.env.VITE_API_BASE || "http://localhost:8000"}${item.videoUrl || item.imageUrl}&download=1`;
const link = document.createElement('a');
link.href = url;
link.download = '';
document.body.appendChild(link);
link.click();
document.body.removeChild(link);
}}
>
<DownloadOutlined style={{ color: '#fff', fontSize: 14 }} />
</div>
</Tooltip>
</div>
</>
)}
{isSelectionMode && isSelectedItem && (
<div style={{
position: 'absolute',
top: 0,
left: 0,
right: 0,
bottom: 0,
border: '3px solid #10b981',
borderRadius: 4,
pointerEvents: 'none',
zIndex: 5,
}} />
)}
</div>
);
})}
bottom: 0,
border: '3px solid #10b981',
borderRadius: 4,
pointerEvents: 'none',
zIndex: 5,
}} />
)}
</div>
);
})}
</div>
{/* 分组内加载更多 */}
{group.total && group.total > group.items.length && (
@@ -1545,7 +1628,7 @@ const GeneratedRecord: React.FC = () => {
},
});
}}
style={{ borderRadius: 8, background: 'linear-gradient(135deg, #bf0b0a 0%, #ff8165 100%)', border: 'none', boxShadow: '0 4px 16px rgba(191, 11, 10, 0.35)', fontWeight: 600, padding: '8px 24px' }}
style={{ borderRadius: 8, background: 'linear-gradient(135deg, #bf0b0a 0%, #ff8165 100%)', fontWeight: 600, padding: '8px 24px' }}
disabled={isMediaExpired(previewItem.videoUrl || previewItem.imageUrl)}
>
删除