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

Merge branch 'main' of gitee.com:wg123/video-gen

Browse Source
This commit is contained in:
Lrd
2026-07-03 13:15:50 +08:00
parent 741b138a22 9ba24a4f51
commit 30db9fb425
8 changed files with 139 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
video-gen-admin/dist/assets/index-BhcMFup1.js → video-gen-admin/dist/assets/index-kmFIWx83.js
Vendored
+1 -1
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>
<script type="module" crossorigin src="/assets/index-BhcMFup1.js"></script> <script type="module" crossorigin src="/assets/index-kmFIWx83.js"></script>
<link rel="stylesheet" crossorigin href="/assets/index-D7ShJUt4.css"> <link rel="stylesheet" crossorigin href="/assets/index-D7ShJUt4.css">
</head> </head>
<body> <body>
video-gen-admin/src/pages/AdminGenerationAiRecords.tsx
+40 -13
View File
@@ -15,6 +15,7 @@ import {
Typography, Typography,
} from 'antd'; } from 'antd';
import { import {
AudioOutlined,
CheckCircleOutlined, CheckCircleOutlined,
ClockCircleOutlined, ClockCircleOutlined,
CloseCircleOutlined, CloseCircleOutlined,
@@ -258,8 +259,8 @@ const AdminGenerationAiRecords: React.FC = () => {
userId: queryUserId || undefined, userId: queryUserId || undefined,
userName: queryUserName || undefined, userName: queryUserName || undefined,
engineId: queryEngineId || undefined, engineId: queryEngineId || undefined,
createdStart: queryCreatedRange?.[0]?.toISOString?.(), createdStart: queryCreatedRange?.[0]?.format?.('YYYY-MM-DDTHH:mm:ss'),
createdEnd: queryCreatedRange?.[1]?.toISOString?.(), createdEnd: queryCreatedRange?.[1]?.format?.('YYYY-MM-DDTHH:mm:ss'),
page, page,
pageSize: PAGE_SIZE, pageSize: PAGE_SIZE,
}); });
@@ -436,18 +437,37 @@ const AdminGenerationAiRecords: React.FC = () => {
}, },
}, },
{ {
title: '附件', key: 'references', width: 110, title: '附件', key: 'references', width: 100,
render: (_: any, r: GenerationAITaskOut) => { render: (_: any, r: GenerationAITaskOut) => {
const refs = r.mediaReferences || []; const refs = r.mediaReferences || [];
if (refs.length === 0) return <Typography.Text style={{ fontSize: 12, color: '#94a3b8' }}>无</Typography.Text>; if (refs.length === 0) return <Typography.Text style={{ fontSize: 12, color: '#94a3b8' }}>无</Typography.Text>;
const imgCount = refs.filter((ref) => getReferenceType(ref) === 'image').length;
const vidCount = refs.filter((ref) => getReferenceType(ref) === 'video').length;
const otherCount = refs.length - imgCount - vidCount;
return ( return (
<Space size={4} wrap> <Space size={2} wrap>
{imgCount > 0 ? <Tag color="purple" icon={<FileImageOutlined />}>{imgCount} 图</Tag> : null} {refs.map((ref, idx) => {
{vidCount > 0 ? <Tag color="geekblue" icon={<VideoCameraOutlined />}>{vidCount} 视频</Tag> : null} const refUrl = getReferenceUrl(ref);
{otherCount > 0 ? <Tag>{otherCount} 文件</Tag> : null} const refType = getReferenceType(ref);
const title = typeof ref.name === 'string' && ref.name ? ref.name : `附件 ${idx + 1}`;
if (!refUrl) return null;
const icon = refType === 'video'
? <VideoCameraOutlined style={{ color: '#6366f1' }} />
: refType === 'audio'
? <AudioOutlined style={{ color: '#f59e0b' }} />
: <FileImageOutlined style={{ color: '#8b5cf6' }} />;
return (
<Tooltip key={idx} title={title}>
<Button
size="small"
type="text"
icon={icon}
style={{ width: 28, height: 28, padding: 0 }}
onClick={(e) => {
e.stopPropagation();
handlePreviewResource(refUrl, refType === 'video' ? 'video' : 'image', title);
}}
/>
</Tooltip>
);
})}
</Space> </Space>
); );
}, },
@@ -465,7 +485,7 @@ const AdminGenerationAiRecords: React.FC = () => {
icon={<PlayCircleOutlined />} icon={<PlayCircleOutlined />}
type="link" type="link"
style={{ padding: 0 }} style={{ padding: 0 }}
onClick={(e) => { e.stopPropagation(); handleOpenExternalResource(r.videoUrl, '视频'); }} onClick={(e) => { e.stopPropagation(); handlePreviewResource(r.videoUrl!, 'video', '生成视频'); }}
> >
看视频 看视频
</Button> </Button>
@@ -478,7 +498,7 @@ const AdminGenerationAiRecords: React.FC = () => {
icon={<FileImageOutlined />} icon={<FileImageOutlined />}
type="link" type="link"
style={{ padding: 0 }} style={{ padding: 0 }}
onClick={(e) => { e.stopPropagation(); handleOpenExternalResource(r.imageUrl, '图片'); }} onClick={(e) => { e.stopPropagation(); handlePreviewResource(r.imageUrl!, 'image', '生成图片'); }}
> >
看图片 看图片
</Button> </Button>
@@ -945,7 +965,14 @@ const AdminGenerationAiRecords: React.FC = () => {
/> />
<RangePicker <RangePicker
value={createdRange} value={createdRange}
onChange={(dates) => { setCreatedRange(dates); }} onChange={(dates) => {
if (dates && dates[0] && dates[1]) {
// 自动将结束时间补到当天 23:59:59，确保选一天也能看到整天数据
setCreatedRange([dates[0].startOf('day'), dates[1].endOf('day')]);
} else {
setCreatedRange(dates);
}
}}
placeholder={['开始日期', '结束日期']} placeholder={['开始日期', '结束日期']}
/> />
<Input <Input
video-gen-api/app/services/generation_recovery_service.py
+48 -11
View File
@@ -446,7 +446,12 @@ async def recover_one_generation_task(
message=f"{source} 发现任务已到 deadline 且存在供应商任务ID，投递 poll 队列做最终查询", message=f"{source} 发现任务已到 deadline 且存在供应商任务ID，投递 poll 队列做最终查询",
detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload}, detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload},
) )
poll_generation_task.apply_async(args=[task.id], queue=POLL_QUEUE, countdown=0) poll_generation_task.apply_async(
args=[task.id],
kwargs={"force_due": True},
queue=POLL_QUEUE,
countdown=0,
)
await register_poll_active( await register_poll_active(
task, task,
check_at=_poll_queue_timeout_at(), check_at=_poll_queue_timeout_at(),
@@ -487,16 +492,30 @@ async def recover_one_generation_task(
) )
return "skip_video_poll_not_due" return "skip_video_poll_not_due"
original_next_poll_at = ensure_aware_utc(task.next_poll_at)
queue_hold_until = _poll_queue_timeout_at(current_time)
task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value
task.next_poll_at = _poll_queue_timeout_at(current_time) # 这里仍复用 next_poll_at 做短暂队列保护，避免启动容灾重复投递。
# 真正消费时通过 force_due=True 跳过“未到期”校验，避免保护时间反向阻塞本次 poll。
task.next_poll_at = queue_hold_until
await db.commit() await db.commit()
await log_task_event( await log_task_event(
task, task,
event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value, event_type=ChatGenerationTaskEventType.GENERATION_RECOVERY_ENQUEUE.value,
message=f"{source} 发现任务存在供应商任务ID，恢复投递轮询队列", message=f"{source} 发现任务存在供应商任务ID，恢复投递轮询队列",
detail={"pipeline_stage": task.pipeline_stage, "payload": redis_payload}, detail={
"pipeline_stage": task.pipeline_stage,
"payload": redis_payload,
"due_next_poll_at": original_next_poll_at,
"queue_hold_until": queue_hold_until,
},
)
poll_generation_task.apply_async(
args=[task.id],
kwargs={"force_due": True},
queue=POLL_QUEUE,
countdown=0,
) )
poll_generation_task.apply_async(args=[task.id], queue=POLL_QUEUE, countdown=0)
await register_poll_active( await register_poll_active(
task, task,
check_at=task.next_poll_at, check_at=task.next_poll_at,
@@ -696,6 +715,8 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]:
results: dict[str, int] = {} results: dict[str, int] = {}
dispatched_task_ids: list[str] = [] dispatched_task_ids: list[str] = []
dispatched_due_next_poll_at_by_id: dict[str, datetime | None] = {}
dispatched_queue_hold_until_by_id: dict[str, datetime] = {}
for task in tasks: for task in tasks:
action = "skip_unknown" action = "skip_unknown"
@@ -723,9 +744,14 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]:
action = "skip_no_provider_task_id" action = "skip_no_provider_task_id"
continue continue
original_next_poll_at = ensure_aware_utc(task.next_poll_at)
queue_hold_until = _poll_queue_timeout_at(current_time)
task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value task.pipeline_stage = ChatGenerationPipelineStage.WAITING_REMOTE.value
# 设置一个队列消费保护时间，避免 Beat 下一分钟看到旧 next_poll_at 又重复投递。 # 设置一个队列消费保护时间，避免 Beat 下一分钟看到旧 next_poll_at 又重复投递。
task.next_poll_at = _poll_queue_timeout_at(current_time) # poll worker 会通过 force_due=True 消费本次到期任务，避免该保护时间被误判为业务未到期。
task.next_poll_at = queue_hold_until
dispatched_due_next_poll_at_by_id[task.id] = original_next_poll_at
dispatched_queue_hold_until_by_id[task.id] = queue_hold_until
dispatched_task_ids.append(task.id) dispatched_task_ids.append(task.id)
action = "dispatch_poll" action = "dispatch_poll"
except Exception as exc: except Exception as exc:
@@ -745,30 +771,41 @@ async def dispatch_due_poll_tasks_once(db: AsyncSession) -> dict[str, Any]:
if dispatched_task_ids: if dispatched_task_ids:
fresh_result = await db.execute( fresh_result = await db.execute(
select(ChatGenerationTask) select(ChatGenerationTask)
.where( .where(
ChatGenerationTask.id.in_(dispatched_task_ids), ChatGenerationTask.id.in_(dispatched_task_ids),
ChatGenerationTask.deleted_at.is_(None), ChatGenerationTask.deleted_at.is_(None),
) )
.execution_options(populate_existing=True) .execution_options(populate_existing=True)
) )
fresh_tasks = fresh_result.scalars().all() fresh_tasks = fresh_result.scalars().all()
enqueued_count = 0 enqueued_count = 0
for task in fresh_tasks: for task in fresh_tasks:
queue_hold_until = dispatched_queue_hold_until_by_id.get(task.id) or ensure_aware_utc(task.next_poll_at) or _poll_queue_timeout_at(current_time)
due_next_poll_at = dispatched_due_next_poll_at_by_id.get(task.id)
await register_poll_active( await register_poll_active(
task, task,
check_at=task.next_poll_at or _poll_queue_timeout_at(current_time), check_at=queue_hold_until,
next_poll_at=task.next_poll_at, next_poll_at=queue_hold_until,
reason="due_dispatch_poll_queued", reason="due_dispatch_poll_queued",
) )
await log_task_event( await log_task_event(
task, task,
event_type=ChatGenerationTaskEventType.POLL_DISPATCH_DUE.value, event_type=ChatGenerationTaskEventType.POLL_DISPATCH_DUE.value,
message="视频 next_poll_at 到期，已投递 provider poll 队列", message="视频 next_poll_at 到期，已投递 provider poll 队列",
detail={"next_poll_at": task.next_poll_at, "queue": POLL_QUEUE}, detail={
"due_next_poll_at": due_next_poll_at,
"queue_hold_until": queue_hold_until,
"queue": POLL_QUEUE,
},
)
poll_generation_task.apply_async(
args=[task.id],
kwargs={"force_due": True},
queue=POLL_QUEUE,
countdown=0,
) )
poll_generation_task.apply_async(args=[task.id], queue=POLL_QUEUE, countdown=0)
enqueued_count += 1 enqueued_count += 1
# 如果 log_task_event 内部不 commit，这里要提交一次 # 如果 log_task_event 内部不 commit，这里要提交一次
video-gen-api/app/services/user_oauth_service.py
+4 -4
View File
@@ -1,5 +1,5 @@
import random import random
from datetime import datetime, timedelta from datetime import datetime, timedelta, timezone
import httpx import httpx
from sqlalchemy import select, func from sqlalchemy import select, func
@@ -136,8 +136,8 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int,
data = data.get("data", {}) data = data.get("data", {})
access_token = data.get("access_token", "") access_token = data.get("access_token", "")
refresh_token = data.get("refresh_token", "") refresh_token = data.get("refresh_token", "")
expires_in = datetime.now() + timedelta(seconds=data.get("expires_in", 0)) expires_in = datetime.now(timezone.utc) + timedelta(seconds=data.get("expires_in", 0))
refresh_token_expires_in = datetime.now() + timedelta(seconds=data.get("refresh_token_expires_in", 0)) refresh_token_expires_in = datetime.now(timezone.utc) + timedelta(seconds=data.get("refresh_token_expires_in", 0))
#2.获取已授权角色账户，一个授权可能有多个角色账户 #2.获取已授权角色账户，一个授权可能有多个角色账户
url = "https://api.oceanengine.com/open_api/oauth2/advertiser/get/" url = "https://api.oceanengine.com/open_api/oauth2/advertiser/get/"
response = await client.get( response = await client.get(
@@ -222,7 +222,7 @@ async def get_juliang_token(app_id: str, secret: str, code: str, open_type: int,
# 1. 软删除已消失的账户 # 1. 软删除已消失的账户
for account_id in old_account_ids - new_account_ids: for account_id in old_account_ids - new_account_ids:
existing_accounts[account_id].deleted_at = datetime.now() existing_accounts[account_id].deleted_at = datetime.now(timezone.utc)
# 2. 更新或新增账户 # 2. 更新或新增账户
for account in account_list: for account in account_list:
video-gen-api/app/tasks/generation_poll_tasks.py
+7 -4
View File
@@ -271,7 +271,7 @@ async def _schedule_next_poll(
) )
async def _run(task_id: str): async def _run(task_id: str, *, force_due: bool = False):
async with async_session() as db: async with async_session() as db:
result = await db.execute(select(ChatGenerationTask).where( result = await db.execute(select(ChatGenerationTask).where(
ChatGenerationTask.id == task_id, ChatGenerationTask.id == task_id,
@@ -296,7 +296,10 @@ async def _run(task_id: str):
current_time = _now() current_time = _now()
if is_video_generation_task(task): if is_video_generation_task(task):
ensure_video_poll_fields(task, now=current_time) ensure_video_poll_fields(task, now=current_time)
if is_poll_not_due(task, now=current_time): # dispatcher / recovery 已经在投递前确认到期时，会传 force_due=True。
# 这样可以避免投递侧为了防重复消费临时写入的 next_poll_at，
# 又被当前 worker 当成“业务下一次轮询时间”而误判未到期。
if not force_due and is_poll_not_due(task, now=current_time):
await db.commit() await db.commit()
await _skip_not_due(task) await _skip_not_due(task)
return return
@@ -481,9 +484,9 @@ async def _run(task_id: str):
if celery_app: if celery_app:
@celery_app.task(name="generation.poll_generation_task", bind=True, max_retries=3, default_retry_delay=30) @celery_app.task(name="generation.poll_generation_task", bind=True, max_retries=3, default_retry_delay=30)
def poll_generation_task(self, task_id: str): def poll_generation_task(self, task_id: str, force_due: bool = False):
try: try:
return run_async(_run(task_id)) return run_async(_run(task_id, force_due=bool(force_due)))
except Exception as exc: except Exception as exc:
# 只重试基础设施异常；供应商失败/业务失败已在 _run 内处理。 # 只重试基础设施异常；供应商失败/业务失败已在 _run 内处理。
retries = int(getattr(self.request, "retries", 0) or 0) + 1 retries = int(getattr(self.request, "retries", 0) or 0) + 1
video-gen-api/app/tasks/token_refresh_task.py
+2 -2
View File
@@ -93,8 +93,8 @@ async def refresh_juliang_token(oauth: UserOAuth, app: UserOAuthApp, db: AsyncSe
data = data.get("data", {}) data = data.get("data", {})
new_access_token = data.get("access_token", "") new_access_token = data.get("access_token", "")
new_refresh_token = data.get("refresh_token", "") new_refresh_token = data.get("refresh_token", "")
expires_in = datetime.now(tz=oauth.access_token_expired.tzinfo) + timedelta(seconds=data.get("expires_in", 0)) expires_in = datetime.now(timezone.utc) + 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)) refresh_token_expires_in = datetime.now(timezone.utc) + timedelta(seconds=data.get("refresh_token_expires_in", 0))
from sqlalchemy import update from sqlalchemy import update
video-gen-api/app/utils/douyinRequest.py
+1 -1
View File
@@ -51,7 +51,7 @@ class DouyinRequest:
token = cache.get("token") token = cache.get("token")
expired_at_str = cache.get("expired_at") expired_at_str = cache.get("expired_at")
if token and expired_at_str: if token and expired_at_str:
expired_at = datetime.fromisoformat(expired_at_str) expired_at = datetime.fromisoformat(expired_at_str).replace(tzinfo=timezone.utc)
if expired_at > datetime.now(timezone.utc): if expired_at > datetime.now(timezone.utc):
return token return token
except Exception as e: except Exception as e: