diff --git a/app/admin/asr/videos/page.tsx b/app/admin/asr/videos/page.tsx index 2bfd283..7c461b8 100644 --- a/app/admin/asr/videos/page.tsx +++ b/app/admin/asr/videos/page.tsx @@ -1,8 +1,9 @@ "use client"; +import { Pause, Play, RefreshCw, Square } from "lucide-react"; import Image from "next/image"; import Link from "next/link"; -import { useCallback, useEffect, useState } from "react"; +import { useCallback, useEffect, useMemo, useState } from "react"; import BackButton from "@/app/components/BackButton"; type VideoTranscript = { @@ -33,24 +34,168 @@ type ApiResponse = { total: number; }; +type Filter = "all" | "transcribed" | "pending" | "tagged" | "missingTags"; +type BatchScope = + | "missingTags" + | "pending" + | "transcribed" + | "tagged" + | "all" + | "selected"; +type BatchJobStatus = + | "queued" + | "running" + | "paused" + | "completed" + | "canceled" + | "failed"; +type BatchItemStatus = + | "pending" + | "running" + | "success" + | "failed" + | "canceled"; + +type BatchJobItem = { + index: number; + awemeId: string; + desc: string; + status: BatchItemStatus; + tags: string[]; + error: string | null; + startedAt: string | null; + finishedAt: string | null; + durationMs: number | null; +}; + +type BatchJob = { + id: string; + status: BatchJobStatus; + scope: BatchScope; + selection: { + sourceCount: number; + matchedCount: number; + selectedCount: number; + rangeStart: number; + rangeEnd: number; + requestedIdsCount: number; + }; + progress: { + total: number; + completed: number; + failed: number; + canceled: number; + pending: number; + running: number; + processed: number; + percent: number; + }; + currentItem: BatchJobItem | null; + pauseRequested: boolean; + cancelRequested: boolean; + error: string | null; + logs: { at: string; message: string; awemeId: string | null }[]; + createdAt: string; + startedAt: string | null; + updatedAt: string; + finishedAt: string | null; + items?: BatchJobItem[]; +}; + +type JobsResponse = { + jobs: BatchJob[]; + activeJob: BatchJob | null; +}; + +type JobResponse = { + job: BatchJob; +}; + +const scopeOptions: { value: BatchScope; label: string }[] = [ + { value: "missingTags", label: "缺标签视频" }, + { value: "pending", label: "待转写视频" }, + { value: "transcribed", label: "已转写视频" }, + { value: "tagged", label: "已有标签视频" }, + { value: "all", label: "全部视频" }, + { value: "selected", label: "勾选的视频" }, +]; + +const statusLabels: Record = { + queued: "排队中", + running: "执行中", + paused: "已暂停", + completed: "已完成", + canceled: "已取消", + failed: "失败", +}; + +const itemStatusLabels: Record = { + pending: "等待", + running: "处理中", + success: "成功", + failed: "失败", + canceled: "已取消", +}; + +const terminalJobStatuses = new Set([ + "completed", + "canceled", + "failed", +]); + +function hasContentTags(video: Video) { + return video.content_tags.length > 0; +} + +function matchVideoScope( + video: Video, + scope: BatchScope, + selectedIds: Set, +) { + if (scope === "all") return true; + if (scope === "selected") return selectedIds.has(video.aweme_id); + if (scope === "missingTags") return !hasContentTags(video); + if (scope === "pending") return video.transcript === null; + if (scope === "transcribed") return video.transcript !== null; + if (scope === "tagged") return hasContentTags(video); + return true; +} + +function isTerminalJob(job: BatchJob | null) { + return !job || terminalJobStatuses.has(job.status); +} + +function parsePositiveInt(value: string, fallback: number) { + const parsed = Number.parseInt(value, 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; +} + +async function readApiError(response: Response, fallback: string) { + const payload = (await response.json().catch(() => null)) as { + error?: string; + } | null; + return payload?.error ?? fallback; +} + +function formatTime(value: string | null) { + if (!value) return "-"; + return new Date(value).toLocaleString("zh-CN", { hour12: false }); +} + export default function AsrVideosPage() { const [videos, setVideos] = useState([]); const [loading, setLoading] = useState(true); const [error, setError] = useState(null); const [transcribing, setTranscribing] = useState>(new Set()); - const [filter, setFilter] = useState< - "all" | "transcribed" | "pending" | "tagged" | "missingTags" - >("all"); - const [batchTranscribing, setBatchTranscribing] = useState(false); - const [batchProgress, setBatchProgress] = useState({ - current: 0, - total: 0, - completed: 0, - failed: 0, - currentAwemeId: null as string | null, - currentDesc: null as string | null, - }); - const [batchMessage, setBatchMessage] = useState(null); + const [filter, setFilter] = useState("all"); + const [selectedIds, setSelectedIds] = useState>(new Set()); + const [taskScope, setTaskScope] = useState("missingTags"); + const [rangeStart, setRangeStart] = useState("1"); + const [rangeEnd, setRangeEnd] = useState(""); + const [job, setJob] = useState(null); + const [jobHistory, setJobHistory] = useState([]); + const [jobLoading, setJobLoading] = useState(false); + const [jobError, setJobError] = useState(null); const fetchVideos = useCallback(async (showLoading = true) => { try { @@ -61,6 +206,7 @@ export default function AsrVideosPage() { } const data: ApiResponse = await response.json(); setVideos(data.videos); + setError(null); } catch (err) { setError(err instanceof Error ? err.message : "Unknown error"); } finally { @@ -68,22 +214,68 @@ export default function AsrVideosPage() { } }, []); + const refreshJob = useCallback(async (jobId: string) => { + const response = await fetch(`/api/admin/asr/jobs/${jobId}`); + if (!response.ok) { + throw new Error(await readApiError(response, "获取任务状态失败")); + } + const data: JobResponse = await response.json(); + setJob(data.job); + setJobError(null); + return data.job; + }, []); + + const fetchJobs = useCallback(async () => { + try { + const response = await fetch("/api/admin/asr/jobs"); + if (!response.ok) { + throw new Error(await readApiError(response, "获取任务列表失败")); + } + const data: JobsResponse = await response.json(); + setJobHistory(data.jobs); + const latestJob = data.activeJob ?? data.jobs[0] ?? null; + if (latestJob) { + await refreshJob(latestJob.id); + } else { + setJob(null); + } + setJobError(null); + } catch (err) { + setJobError(err instanceof Error ? err.message : "获取任务列表失败"); + } + }, [refreshJob]); + useEffect(() => { fetchVideos(); - }, [fetchVideos]); + fetchJobs(); + }, [fetchJobs, fetchVideos]); + + useEffect(() => { + if (!job || terminalJobStatuses.has(job.status)) return; + + const intervalId = window.setInterval(async () => { + try { + const nextJob = await refreshJob(job.id); + if (terminalJobStatuses.has(nextJob.status)) { + await fetchVideos(false); + await fetchJobs(); + } + } catch (err) { + setJobError(err instanceof Error ? err.message : "刷新任务状态失败"); + } + }, 2000); + + return () => window.clearInterval(intervalId); + }, [fetchJobs, fetchVideos, job, refreshJob]); - // 简单的 sleep 工具 const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); - const hasContentTags = (video: Video) => video.content_tags.length > 0; - type PollOptions = { intervalMs?: number; maxAttempts?: number; }; - // 轮询某个视频的识别和标签状态直至完成或超时 const pollVideoUntilTranscribed = async ( awemeId: string, { intervalMs = 2000, maxAttempts = 300 }: PollOptions = {}, @@ -94,13 +286,13 @@ export default function AsrVideosPage() { if (resp.ok) { const data: ApiResponse = await resp.json(); setVideos(data.videos); - const v = data.videos.find((x) => x.aweme_id === awemeId); - if ((v?.content_tags.length ?? 0) > 0) { + const video = data.videos.find((item) => item.aweme_id === awemeId); + if ((video?.content_tags.length ?? 0) > 0) { return true; } } } catch { - // 忽略网络错误，继续轮询 + // 网络抖动时继续轮询 } await sleep(intervalMs); } @@ -116,12 +308,8 @@ export default function AsrVideosPage() { if (!response.ok) { throw new Error("Failed to transcribe video"); } - // 轮询直到该视频转写完成或超时，避免后端异步导致立即刷新拿不到最新状态 const done = await pollVideoUntilTranscribed(awemeId); - if (!done) { - // 兜底刷新一次 - await fetchVideos(false); - } + if (!done) await fetchVideos(false); } catch (err) { alert(err instanceof Error ? err.message : "Transcription failed"); } finally { @@ -133,103 +321,162 @@ export default function AsrVideosPage() { } }; - const handleBatchTranscribe = async () => { - const pendingVideos = videos.filter((v) => !hasContentTags(v)); - if (pendingVideos.length === 0) { - alert("没有缺失标签的视频"); - return; - } + const filteredVideos = useMemo( + () => + videos.filter((video) => { + if (filter === "transcribed") return video.transcript !== null; + if (filter === "pending") return video.transcript === null; + if (filter === "tagged") return hasContentTags(video); + if (filter === "missingTags") return !hasContentTags(video); + return true; + }), + [filter, videos], + ); + const stats = useMemo( + () => ({ + total: videos.length, + transcribed: videos.filter((video) => video.transcript !== null).length, + pending: videos.filter((video) => video.transcript === null).length, + tagged: videos.filter((video) => hasContentTags(video)).length, + missingTags: videos.filter((video) => !hasContentTags(video)).length, + }), + [videos], + ); + + const taskCandidates = useMemo( + () => + videos.filter((video) => matchVideoScope(video, taskScope, selectedIds)), + [selectedIds, taskScope, videos], + ); + const startIndex = parsePositiveInt(rangeStart, 1); + const endIndex = rangeEnd.trim() + ? parsePositiveInt(rangeEnd, taskCandidates.length) + : taskCandidates.length; + const estimatedCount = Math.max( + 0, + taskCandidates.slice(startIndex - 1, Math.max(startIndex, endIndex)).length, + ); + const selectedVisible = + filteredVideos.length > 0 && + filteredVideos.every((video) => selectedIds.has(video.aweme_id)); + const activeJobRunning = job ? !isTerminalJob(job) : false; + const canCreateJob = + !activeJobRunning && + estimatedCount > 0 && + (taskScope !== "selected" || selectedIds.size > 0); + + const toggleVideoSelection = useCallback((awemeId: string) => { + setSelectedIds((current) => { + const next = new Set(current); + if (next.has(awemeId)) next.delete(awemeId); + else next.add(awemeId); + return next; + }); + }, []); + + const toggleVisibleSelection = useCallback(() => { + setSelectedIds((current) => { + const next = new Set(current); + if (filteredVideos.every((video) => next.has(video.aweme_id))) { + for (const video of filteredVideos) next.delete(video.aweme_id); + } else { + for (const video of filteredVideos) next.add(video.aweme_id); + } + return next; + }); + }, [filteredVideos]); + + const handleCreateBatchJob = async () => { + if (!canCreateJob) return; + const scopeLabel = + scopeOptions.find((option) => option.value === taskScope)?.label ?? + "当前范围"; if ( !confirm( - `确定要为 ${pendingVideos.length} 个视频生成 ASR 标签吗？这会重新执行识别流程，可能需要较长时间。`, + `确定发布后端批量 ASR 任务吗？\n范围：${scopeLabel}\n预计处理：${estimatedCount} 个视频`, ) ) { return; } - setBatchTranscribing(true); - setBatchMessage(null); - setBatchProgress({ - current: 0, - total: pendingVideos.length, - completed: 0, - failed: 0, - currentAwemeId: null, - currentDesc: null, - }); - - for (let i = 0; i < pendingVideos.length; i++) { - const video = pendingVideos[i]; - setBatchProgress((prev) => ({ - ...prev, - current: i + 1, - currentAwemeId: video.aweme_id, - currentDesc: video.desc || "无描述", - })); - setTranscribing((prev) => new Set(prev).add(video.aweme_id)); - - try { - const response = await fetch( - `/api/asr?awemeId=${encodeURIComponent(video.aweme_id)}&async=1`, - ); - if (!response.ok) { - throw new Error(`Failed to start ASR for ${video.aweme_id}`); - } - - const done = await pollVideoUntilTranscribed(video.aweme_id); - setBatchProgress((prev) => ({ - ...prev, - completed: done ? prev.completed + 1 : prev.completed, - failed: done ? prev.failed : prev.failed + 1, - })); - } catch (err) { - console.error(`Error transcribing video ${video.aweme_id}:`, err); - setBatchProgress((prev) => ({ - ...prev, - failed: prev.failed + 1, - })); - } finally { - setTranscribing((prev) => { - const next = new Set(prev); - next.delete(video.aweme_id); - return next; - }); + setJobLoading(true); + setJobError(null); + try { + const response = await fetch("/api/admin/asr/jobs", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + scope: taskScope, + rangeStart: startIndex, + ...(rangeEnd.trim() ? { rangeEnd: endIndex } : {}), + ...(taskScope === "selected" ? { awemeIds: [...selectedIds] } : {}), + }), + }); + if (!response.ok) { + throw new Error(await readApiError(response, "发布任务失败")); } + const data: JobResponse = await response.json(); + setJob(data.job); + setJobHistory((current) => [ + data.job, + ...current.filter((item) => item.id !== data.job.id), + ]); + } catch (err) { + setJobError(err instanceof Error ? err.message : "发布任务失败"); + } finally { + setJobLoading(false); } - - await fetchVideos(false); - setBatchTranscribing(false); - setBatchMessage("缺失标签生成流程已结束"); }; - const filteredVideos = videos.filter((v) => { - if (filter === "transcribed") return v.transcript !== null; - if (filter === "pending") return v.transcript === null; - if (filter === "tagged") return hasContentTags(v); - if (filter === "missingTags") return !hasContentTags(v); - return true; - }); + const handleControlJob = async (action: "pause" | "resume" | "cancel") => { + if (!job) return; + if (action === "cancel" && !confirm("确定取消当前批量任务吗？")) return; - const stats = { - total: videos.length, - transcribed: videos.filter((v) => v.transcript !== null).length, - pending: videos.filter((v) => v.transcript === null).length, - tagged: videos.filter((v) => hasContentTags(v)).length, - missingTags: videos.filter((v) => !hasContentTags(v)).length, + setJobLoading(true); + setJobError(null); + try { + const response = await fetch(`/api/admin/asr/jobs/${job.id}`, { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ action }), + }); + if (!response.ok) { + throw new Error(await readApiError(response, "操作任务失败")); + } + const data: JobResponse = await response.json(); + setJob(data.job); + } catch (err) { + setJobError(err instanceof Error ? err.message : "操作任务失败"); + } finally { + setJobLoading(false); + } }; - const processedCount = batchProgress.completed + batchProgress.failed; - const progressPercent = - batchProgress.total > 0 - ? Math.round((processedCount / batchProgress.total) * 100) - : 0; + + const handleSelectHistoryJob = async (jobId: string) => { + if (!jobId) return; + setJobLoading(true); + try { + await refreshJob(jobId); + } catch (err) { + setJobError(err instanceof Error ? err.message : "获取任务状态失败"); + } finally { + setJobLoading(false); + } + }; + + const recentItems = (job?.items ?? []) + .filter((item) => item.status !== "pending") + .slice(-6) + .reverse(); + const recentLogs = (job?.logs ?? []).slice(-8).reverse(); if (loading) { return (
-
+

加载中...

@@ -259,7 +506,6 @@ export default function AsrVideosPage() { return (
- {/* Header */}

@@ -268,7 +514,6 @@ export default function AsrVideosPage() {

管理视频的语音识别和自动标签状态

- {/* Stats */}
总视频数
@@ -299,131 +544,373 @@ export default function AsrVideosPage() {
- {/* Filter */}
-
-
-
- - - - - +
+
+ + + + + +
+ +
+ 已勾选 {selectedIds.size} 个视频 +
+
+
+ +
+
+
+
+

+ 后端批量任务 +

+

+ 发布后由服务端串行执行，页面刷新后仍可查询当前进度。 +

-
- {batchTranscribing && ( -
- 正在生成: {batchProgress.current} / {batchProgress.total} -
+
+ + + +
+ +
+ 范围内匹配 {taskCandidates.length} 个 + 本次预计处理 {estimatedCount} 个 + {taskScope === "selected" && selectedIds.size === 0 && ( + 请先勾选视频 )} +
+ + {jobError && ( +
+ {jobError} +
+ )} + +
+
- {(batchTranscribing || batchMessage) && ( -
-
- - {batchTranscribing ? "正在生成缺失标签" : batchMessage} - - - {processedCount} / {batchProgress.total} - -
-
-
-
-
- - 已完成 {batchProgress.completed}，失败/超时{" "} - {batchProgress.failed} - - {batchTranscribing && batchProgress.currentAwemeId && ( - - 当前: {batchProgress.currentDesc} ( - {batchProgress.currentAwemeId}) - +
+
+
+
+ {job ? `任务 ${job.id.slice(0, 8)}` : "暂无任务"} +
+ {job && ( +
+ 创建 {formatTime(job.createdAt)}，更新{" "} + {formatTime(job.updatedAt)} +
)}
+ {jobHistory.length > 0 && ( + + )}
- )} + + {job ? ( +
+
+ + {job.cancelRequested + ? "取消中" + : job.pauseRequested + ? "暂停中" + : statusLabels[job.status]} + + + {job.progress.processed} / {job.progress.total} + + + 成功 {job.progress.completed}，失败 {job.progress.failed} + ，取消 {job.progress.canceled} + +
+ +
+
+
+
+
+ 范围 {job.selection.rangeStart}-{job.selection.rangeEnd} + ，匹配 {job.selection.matchedCount}，选中{" "} + {job.selection.selectedCount} +
+
+ + {job.currentItem && ( +
+
+ 当前: {job.currentItem.index}/{job.progress.total} +
+
+ {job.currentItem.desc || job.currentItem.awemeId} +
+
+ )} + +
+ + + + +
+ + {recentItems.length > 0 && ( +
+
+ 最近结果 +
+
+ {recentItems.map((item) => ( +
+ + {item.index}. {item.desc || item.awemeId} + + + {itemStatusLabels[item.status]} + +
+ ))} +
+
+ )} + + {recentLogs.length > 0 && ( +
+
+ 最近日志 +
+
+ {recentLogs.map((log) => ( +
+ {formatTime(log.at)} {log.message} +
+ ))} +
+
+ )} +
+ ) : ( +
+ 还没有批量任务 +
+ )} +
- {/* Video List */}
+ @@ -441,6 +928,15 @@ export default function AsrVideosPage() { {filteredVideos.map((video) => ( + ))} diff --git a/app/api/admin/asr/jobs/[jobId]/route.ts b/app/api/admin/asr/jobs/[jobId]/route.ts new file mode 100644 index 0000000..855880e --- /dev/null +++ b/app/api/admin/asr/jobs/[jobId]/route.ts @@ -0,0 +1,55 @@ +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +import { type NextRequest, NextResponse } from "next/server"; +import { + AsrBatchJobError, + controlAsrBatchJob, + getAsrBatchJob, +} from "@/lib/asr-batch-jobs"; + +type RouteContext = { + params: Promise<{ jobId: string }>; +}; + +export async function GET(_req: NextRequest, { params }: RouteContext) { + const { jobId } = await params; + const job = getAsrBatchJob(jobId, true); + if (!job) { + return NextResponse.json({ error: "任务不存在" }, { status: 404 }); + } + + return NextResponse.json({ job }); +} + +export async function PATCH(req: NextRequest, { params }: RouteContext) { + const { jobId } = await params; + let body: unknown; + try { + body = await req.json(); + } catch { + return NextResponse.json({ error: "Invalid JSON body" }, { status: 400 }); + } + + const action = + body && typeof body === "object" && "action" in body + ? (body as { action?: unknown }).action + : undefined; + + try { + const job = controlAsrBatchJob(jobId, action); + return NextResponse.json({ job }); + } catch (error) { + if (error instanceof AsrBatchJobError) { + return NextResponse.json( + { error: error.message }, + { status: error.status }, + ); + } + console.error("Failed to control ASR batch job:", error); + return NextResponse.json( + { error: "Failed to control ASR batch job" }, + { status: 500 }, + ); + } +} diff --git a/app/api/admin/asr/jobs/route.ts b/app/api/admin/asr/jobs/route.ts new file mode 100644 index 0000000..5594103 --- /dev/null +++ b/app/api/admin/asr/jobs/route.ts @@ -0,0 +1,45 @@ +export const runtime = "nodejs"; +export const dynamic = "force-dynamic"; + +import { type NextRequest, NextResponse } from "next/server"; +import { + AsrBatchJobError, + createAsrBatchJob, + getActiveAsrBatchJob, + listAsrBatchJobs, +} from "@/lib/asr-batch-jobs"; + +export async function GET() { + return NextResponse.json({ + jobs: listAsrBatchJobs(), + activeJob: getActiveAsrBatchJob(), + }); +} + +export async function POST(req: NextRequest) { + let body: unknown; + try { + body = await req.json(); + } catch { + return NextResponse.json({ error: "Invalid JSON body" }, { status: 400 }); + } + + try { + const job = await createAsrBatchJob( + body && typeof body === "object" ? body : {}, + ); + return NextResponse.json({ job }, { status: 201 }); + } catch (error) { + if (error instanceof AsrBatchJobError) { + return NextResponse.json( + { error: error.message }, + { status: error.status }, + ); + } + console.error("Failed to create ASR batch job:", error); + return NextResponse.json( + { error: "Failed to create ASR batch job" }, + { status: 500 }, + ); + } +} diff --git a/app/api/asr/index.ts b/app/api/asr/index.ts index b042e09..428427c 100644 --- a/app/api/asr/index.ts +++ b/app/api/asr/index.ts @@ -4,8 +4,10 @@ import { ASR_CONTENT_TAGS, normalizeAsrContentTags } from "@/lib/asr-tags"; import { downloadFile } from "@/lib/minio"; import { prisma } from "@/lib/prisma"; import { + type ExtractedAudioSegment, type ExtractedVideoFrame, extractAudio, + extractAudioSegments, extractVideoFrames, } from "../media"; import prompt from "./prompt.md"; @@ -16,8 +18,25 @@ type TranscriptAwemeOptions = { log?: AsrLogger; }; +type TranscriptAudioOptions = { + segment?: { + index: number; + total: number; + startMs: number; + endMs: number; + }; +}; + let qwenClient: OpenAI | null = null; +const AUDIO_DATA_URL_PREFIX = "data:;base64,"; +const DEFAULT_QWEN_MAX_STRING_LENGTH = 28_000_000; +const DEFAULT_SAFE_AUDIO_DATA_URL_LENGTH = 24_000_000; +const DEFAULT_AUDIO_BITRATE_KBPS = 128; +const DEFAULT_AUDIO_CHUNK_MS = 10 * 60 * 1000; +const MIN_AUDIO_CHUNK_MS = 60 * 1000; +const CHUNK_REEXTRACT_LIMIT = 5; + function getQwenClient() { if (qwenClient) return qwenClient; @@ -81,11 +100,188 @@ function parseAsrResult(content: string): AsrResult { }); } +function getPositiveIntEnv(name: string, fallback: number) { + const value = Number.parseInt(process.env[name] ?? "", 10); + return Number.isFinite(value) && value > 0 ? value : fallback; +} + +function getAudioBitrateKbps() { + return getPositiveIntEnv( + "ASR_AUDIO_BITRATE_KBPS", + DEFAULT_AUDIO_BITRATE_KBPS, + ); +} + +function getAudioChunkMs() { + return Math.max( + MIN_AUDIO_CHUNK_MS, + getPositiveIntEnv( + "ASR_AUDIO_CHUNK_SECONDS", + DEFAULT_AUDIO_CHUNK_MS / 1000, + ) * 1000, + ); +} + +function getMaxAudioDataUrlLength() { + const providerLimit = getPositiveIntEnv( + "QWEN_MAX_STRING_LENGTH", + DEFAULT_QWEN_MAX_STRING_LENGTH, + ); + const configuredLimit = getPositiveIntEnv( + "ASR_MAX_AUDIO_DATA_URL_LENGTH", + DEFAULT_SAFE_AUDIO_DATA_URL_LENGTH, + ); + return Math.max( + 1_000_000, + Math.min(configuredLimit, providerLimit - 1_000_000), + ); +} + +function getBase64Length(byteLength: number) { + return Math.ceil(byteLength / 3) * 4; +} + +function getAudioDataUrlLength(audio: Buffer) { + return AUDIO_DATA_URL_PREFIX.length + getBase64Length(audio.length); +} + +function estimateAudioDataUrlLength(durationMs: number, bitrateKbps: number) { + const estimatedBytes = Math.ceil((durationMs / 1000) * bitrateKbps * 125); + return AUDIO_DATA_URL_PREFIX.length + getBase64Length(estimatedBytes); +} + +function formatTimestamp(ms: number) { + const totalSeconds = Math.max(0, Math.floor(ms / 1000)); + const hours = Math.floor(totalSeconds / 3600); + const minutes = Math.floor((totalSeconds % 3600) / 60); + const seconds = totalSeconds % 60; + if (hours > 0) { + return `${hours}:${minutes.toString().padStart(2, "0")}:${seconds + .toString() + .padStart(2, "0")}`; + } + return `${minutes}:${seconds.toString().padStart(2, "0")}`; +} + +function getSegmentInstruction(segment?: TranscriptAudioOptions["segment"]) { + if (!segment) return ""; + + return `\n\n当前音频是长视频的第 ${segment.index + 1}/${segment.total} 段，时间范围 ${formatTimestamp(segment.startMs)}-${formatTimestamp(segment.endMs)}。请只转写这一段音频，不要猜测其他分片未出现的台词；content_tags 仍结合整条视频三帧、视频简介和当前分片内容判断。`; +} + +function shouldChunkBeforeFullExtract(durationMs: number) { + if (!durationMs || durationMs <= 0) return false; + return ( + estimateAudioDataUrlLength(durationMs, getAudioBitrateKbps()) > + getMaxAudioDataUrlLength() + ); +} + +function mergeContentTags(results: AsrResult[]) { + const counts = new Map(); + let order = 0; + for (const result of results) { + for (const tag of result.content_tags) { + const current = counts.get(tag); + if (current) { + current.count += 1; + } else { + counts.set(tag, { count: 1, firstSeen: order }); + order += 1; + } + } + } + + const sortedTags = [...counts.entries()].sort( + (left, right) => + right[1].count - left[1].count || left[1].firstSeen - right[1].firstSeen, + ); + const preferredTags = sortedTags.filter(([tag]) => tag !== "其他"); + const tags = (preferredTags.length > 0 ? preferredTags : sortedTags) + .slice(0, 3) + .map(([tag]) => tag); + + return normalizeAsrContentTags(tags); +} + +function mergeLanguages(results: AsrResult[]) { + const languages = results + .flatMap((result) => + (result.language ?? "") + .split(/[,，、\s]+/) + .map((item) => item.trim()) + .filter(Boolean), + ) + .filter((language, index, array) => array.indexOf(language) === index); + + return languages.length > 0 ? languages.join(",") : null; +} + +function getMostCommonAudioType(results: AsrResult[]) { + const counts = new Map(); + let order = 0; + for (const result of results) { + if (!result.audio_type) continue; + const current = counts.get(result.audio_type); + if (current) { + current.count += 1; + } else { + counts.set(result.audio_type, { count: 1, firstSeen: order }); + order += 1; + } + } + + return ( + [...counts.entries()].sort( + (left, right) => + right[1].count - left[1].count || + left[1].firstSeen - right[1].firstSeen, + )[0]?.[0] ?? null + ); +} + +function mergeTranscriptLines(results: AsrResult[]) { + const lines: string[] = []; + for (const result of results) { + for (const line of result.transcript) { + const normalizedLine = line.trim(); + if (!normalizedLine) continue; + if (lines.at(-1) === normalizedLine) continue; + lines.push(normalizedLine); + } + } + return lines; +} + +function mergeAsrResults(results: AsrResult[]): AsrResult { + if (results.length === 0) { + throw new Error("No ASR chunk results returned from Qwen Omni"); + } + + const speechDetected = results.some((result) => result.speech_detected); + const nonSpeechSummaries = results + .map((result) => result.non_speech_summary?.trim()) + .filter((summary): summary is string => Boolean(summary)); + + return AsrSchema.parse({ + speech_detected: speechDetected, + language: speechDetected ? mergeLanguages(results) : null, + audio_type: speechDetected ? null : getMostCommonAudioType(results), + transcript: mergeTranscriptLines(results), + non_speech_summary: + speechDetected || nonSpeechSummaries.length === 0 + ? null + : nonSpeechSummaries.join("；"), + content_tags: mergeContentTags(results), + }); +} + async function transcriptAudio( audio: Buffer, frames: ExtractedVideoFrame[], videoDesc: string, log?: AsrLogger, + options: TranscriptAudioOptions = {}, ) { const base64Audio = Buffer.from(audio).toString("base64"); const client = getQwenClient(); @@ -99,7 +295,7 @@ async function transcriptAudio( })); log?.( - `调用 Qwen Omni: model=${model}, audio=${Math.round(audio.length / 1024)}KB, frames=${frames.length}`, + `调用 Qwen Omni: model=${model}, audio=${Math.round(audio.length / 1024)}KB, audioDataUrl=${getAudioDataUrlLength(audio)}, frames=${frames.length}`, ); const completion = await client.chat.completions.create({ @@ -114,11 +310,14 @@ async function transcriptAudio( })), { type: "input_audio", - input_audio: { data: `data:;base64,${base64Audio}`, format: "mp3" }, + input_audio: { + data: `${AUDIO_DATA_URL_PREFIX}${base64Audio}`, + format: "mp3", + }, }, { type: "text", - text: `${prompt}\n\n视频简介：\n${videoDesc || "无"}\n\n已附带视频三帧，时间点约为：${ + text: `${prompt}${getSegmentInstruction(options.segment)}\n\n视频简介：\n${videoDesc || "无"}\n\n已附带视频三帧，时间点约为：${ frames .map((frame) => `${(frame.timestampMs / 1000).toFixed(1)}s`) .join("、") || "无" @@ -151,6 +350,102 @@ async function transcriptAudio( } } +async function extractSafeAudioSegments( + videoBuffer: Buffer, + durationMs: number, + log?: AsrLogger, +) { + if (!durationMs || durationMs <= 0) { + throw new Error( + "Audio is too large for a single request and video duration is missing", + ); + } + + const maxAudioDataUrlLength = getMaxAudioDataUrlLength(); + let segmentDurationMs = Math.min(getAudioChunkMs(), durationMs); + + for (let attempt = 1; attempt <= CHUNK_REEXTRACT_LIMIT; attempt++) { + log?.( + `提取音频分片: chunk=${Math.round(segmentDurationMs / 1000)}s, bitrate=${getAudioBitrateKbps()}kbps`, + ); + const segments = await extractAudioSegments(videoBuffer, { + durationMs, + segmentDurationMs, + format: "mp3", + bitrateKbps: getAudioBitrateKbps(), + }); + + if (segments.length === 0) { + throw new Error("Failed to extract audio segments from video"); + } + + const largestSegmentLength = Math.max( + ...segments.map((segment) => getAudioDataUrlLength(segment.buffer)), + ); + log?.( + `音频分片提取完成: count=${segments.length}, largestDataUrl=${largestSegmentLength}`, + ); + + if (largestSegmentLength <= maxAudioDataUrlLength) return segments; + + if (segmentDurationMs <= MIN_AUDIO_CHUNK_MS) { + throw new Error( + `Audio segment is still too large after splitting: ${largestSegmentLength}`, + ); + } + + const shrinkRatio = Math.max( + 0.25, + Math.min(0.85, (maxAudioDataUrlLength / largestSegmentLength) * 0.8), + ); + segmentDurationMs = Math.max( + MIN_AUDIO_CHUNK_MS, + Math.floor(segmentDurationMs * shrinkRatio), + ); + log?.( + `音频分片仍过大，缩短分片后重试: ${Math.round(segmentDurationMs / 1000)}s`, + ); + } + + throw new Error("Failed to extract safe audio segments from video"); +} + +async function transcriptAudioSegments( + segments: ExtractedAudioSegment[], + frames: ExtractedVideoFrame[], + videoDesc: string, + log?: AsrLogger, +) { + const results: AsrResult[] = []; + + for (const segment of segments) { + log?.( + `识别音频分片 ${segment.index + 1}/${segments.length}: ${formatTimestamp(segment.startMs)}-${formatTimestamp(segment.endMs)}`, + ); + const result = await transcriptAudio( + segment.buffer, + frames, + videoDesc, + log, + { + segment: { + index: segment.index, + total: segments.length, + startMs: segment.startMs, + endMs: segment.endMs, + }, + }, + ); + results.push(result); + } + + const mergedResult = mergeAsrResults(results); + log?.( + `分片结果合并完成: chunks=${results.length}, transcript=${mergedResult.transcript.length} 段, tags=${mergedResult.content_tags.join("、")}`, + ); + return mergedResult; +} + export async function transcriptAweme( awemeId: string, options: TranscriptAwemeOptions = {}, @@ -179,24 +474,48 @@ export async function transcriptAweme( .join("、")}`, ); - log?.("提取音频"); - const audioDat = await extractAudio(buffer, { - format: "mp3", - bitrateKbps: 128, - }); + let result: AsrResult; + const maxAudioDataUrlLength = getMaxAudioDataUrlLength(); + if (shouldChunkBeforeFullExtract(aweme.duration_ms)) { + log?.( + `预计整段音频会超过安全阈值 ${maxAudioDataUrlLength}，启用长视频分片识别`, + ); + const segments = await extractSafeAudioSegments( + buffer, + aweme.duration_ms, + log, + ); + result = await transcriptAudioSegments(segments, frames, aweme.desc, log); + } else { + log?.("提取音频"); + const audioDat = await extractAudio(buffer, { + format: "mp3", + bitrateKbps: getAudioBitrateKbps(), + }); - if (!audioDat || !audioDat.buffer) { - throw new Error("Failed to extract audio from video"); + if (!audioDat?.buffer) { + throw new Error("Failed to extract audio from video"); + } + + const audioDataUrlLength = getAudioDataUrlLength(audioDat.buffer); + log?.( + `音频提取完成: ${Math.round(audioDat.buffer.length / 1024)}KB, dataUrl=${audioDataUrlLength}`, + ); + + if (audioDataUrlLength > maxAudioDataUrlLength) { + log?.( + `整段音频 data URL 超过安全阈值 ${maxAudioDataUrlLength}，改用分片识别`, + ); + const segments = await extractSafeAudioSegments( + buffer, + aweme.duration_ms, + log, + ); + result = await transcriptAudioSegments(segments, frames, aweme.desc, log); + } else { + result = await transcriptAudio(audioDat.buffer, frames, aweme.desc, log); + } } - log?.(`音频提取完成: ${Math.round(audioDat.buffer.length / 1024)}KB`); - - // 调用 Qwen Omni 生成转写与固定分类标签 - const result = await transcriptAudio( - audioDat.buffer, - frames, - aweme.desc, - log, - ); log?.(`解析完成: tags=${result.content_tags.join("、")}`); // 将转写结果持久化到数据库 diff --git a/app/api/media.ts b/app/api/media.ts index 972c0df..f5c8442 100644 --- a/app/api/media.ts +++ b/app/api/media.ts @@ -82,6 +82,60 @@ export type ExtractedVideoFrame = { timestampMs: number; }; +export type ExtractedAudio = { + buffer: Buffer; + contentType: string; + ext: string; +}; + +export type ExtractedAudioSegment = ExtractedAudio & { + index: number; + startMs: number; + endMs: number; +}; + +type AudioFormat = "mp3" | "aac" | "wav"; + +function resolveAudioEncoding(format: AudioFormat, bitrateKbps: number) { + if (format === "mp3") { + return { + ext: "mp3", + contentType: "audio/mpeg", + codecArgs: ["-c:a", "libmp3lame", "-b:a", `${bitrateKbps}k`], + }; + } + + if (format === "aac") { + return { + ext: "m4a", + contentType: "audio/mp4", + codecArgs: [ + "-c:a", + "aac", + "-b:a", + `${bitrateKbps}k`, + "-movflags", + "+faststart", + ], + }; + } + + return { + ext: "wav", + contentType: "audio/wav", + codecArgs: [ + "-f", + "wav", + "-acodec", + "pcm_s16le", + "-ar", + "44100", + "-ac", + "2", + ], + }; +} + /** * 使用 ffprobe 获取视频时长（毫秒） */ @@ -234,8 +288,8 @@ export async function extractVideoFrames( */ export async function extractAudio( videoBuffer: Buffer, - opts?: { format?: "mp3" | "aac" | "wav"; bitrateKbps?: number }, -): Promise<{ buffer: Buffer; contentType: string; ext: string } | null> { + opts?: { format?: AudioFormat; bitrateKbps?: number }, +): Promise { const ffmpegCmd = process.env.FFMPEG_PATH || "ffmpeg"; const tmpDir = os.tmpdir(); const base = `dy_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; @@ -243,47 +297,8 @@ export async function extractAudio( const format = opts?.format ?? "mp3"; const bitrate = Math.max(32, Math.min(512, opts?.bitrateKbps ?? 192)); // 安全范围 32~512 kbps - - // 根据目标格式设置输出路径、MIME 与编码参数 - let outPath = ""; - let contentType = ""; - let ext = ""; - let codecArgs: string[] = []; - - if (format === "mp3") { - ext = "mp3"; - contentType = "audio/mpeg"; - outPath = path.join(tmpDir, `${base}.${ext}`); - codecArgs = ["-c:a", "libmp3lame", "-b:a", `${bitrate}k`]; - } else if (format === "aac") { - // 使用 m4a 容器更通用 - ext = "m4a"; - contentType = "audio/mp4"; - outPath = path.join(tmpDir, `${base}.${ext}`); - codecArgs = [ - "-c:a", - "aac", - "-b:a", - `${bitrate}k`, - "-movflags", - "+faststart", - ]; - } else { - // wav - ext = "wav"; - contentType = "audio/wav"; - outPath = path.join(tmpDir, `${base}.${ext}`); - codecArgs = [ - "-f", - "wav", - "-acodec", - "pcm_s16le", - "-ar", - "44100", - "-ac", - "2", - ]; - } + const { ext, contentType, codecArgs } = resolveAudioEncoding(format, bitrate); + const outPath = path.join(tmpDir, `${base}.${ext}`); try { await fs.writeFile(inPath, videoBuffer); @@ -320,3 +335,88 @@ export async function extractAudio( } catch {} } } + +/** + * 将视频音轨按时间切成多个音频片段，避免长视频单次请求体过大。 + */ +export async function extractAudioSegments( + videoBuffer: Buffer, + opts: { + durationMs: number; + segmentDurationMs: number; + format?: AudioFormat; + bitrateKbps?: number; + }, +): Promise { + const ffmpegCmd = process.env.FFMPEG_PATH || "ffmpeg"; + const tmpDir = os.tmpdir(); + const base = `dy_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; + const inPath = path.join(tmpDir, `${base}.mp4`); + + const format = opts.format ?? "mp3"; + const bitrate = Math.max(32, Math.min(512, opts.bitrateKbps ?? 128)); + const segmentDurationMs = Math.max(1000, opts.segmentDurationMs); + const durationMs = Math.max(0, opts.durationMs); + const { ext, contentType, codecArgs } = resolveAudioEncoding(format, bitrate); + const outPaths: string[] = []; + + try { + await fs.writeFile(inPath, videoBuffer); + + const segments: ExtractedAudioSegment[] = []; + let index = 0; + for (let startMs = 0; startMs < durationMs; startMs += segmentDurationMs) { + const endMs = Math.min(durationMs, startMs + segmentDurationMs); + if (endMs - startMs < 500) continue; + + const outPath = path.join(tmpDir, `${base}_${index}.${ext}`); + outPaths.push(outPath); + + const args = [ + "-hide_banner", + "-loglevel", + "error", + "-ss", + (startMs / 1000).toFixed(3), + "-t", + ((endMs - startMs) / 1000).toFixed(3), + "-i", + inPath, + "-vn", + ...codecArgs, + "-y", + outPath, + ]; + await execFileAsync(ffmpegCmd, args, { windowsHide: true }); + segments.push({ + buffer: await fs.readFile(outPath), + contentType, + ext, + index, + startMs, + endMs, + }); + index += 1; + } + + return segments; + } catch (error) { + if (isMissingExecutableError(error)) { + console.warn( + "系统未检测到 ffmpeg，可安装并配置 PATH 或设置 FFMPEG_PATH 后启用音频分片提取。", + ); + return []; + } + console.warn(`提取音频分片失败: ${getErrorMessage(error)}`); + throw error; + } finally { + try { + await fs.unlink(inPath); + } catch {} + for (const outPath of outPaths) { + try { + await fs.unlink(outPath); + } catch {} + } + } +} diff --git a/lib/asr-batch-jobs.ts b/lib/asr-batch-jobs.ts new file mode 100644 index 0000000..8efee6b --- /dev/null +++ b/lib/asr-batch-jobs.ts @@ -0,0 +1,563 @@ +import { randomUUID } from "node:crypto"; +import { transcriptAweme } from "@/app/api/asr"; +import { resolveEffectiveContentTags } from "@/lib/asr-tags"; +import { prisma } from "@/lib/prisma"; + +export const ASR_BATCH_SCOPES = [ + "missingTags", + "pending", + "transcribed", + "tagged", + "all", + "selected", +] as const; + +export type AsrBatchScope = (typeof ASR_BATCH_SCOPES)[number]; +export type AsrBatchJobStatus = + | "queued" + | "running" + | "paused" + | "completed" + | "canceled" + | "failed"; +export type AsrBatchItemStatus = + | "pending" + | "running" + | "success" + | "failed" + | "canceled"; +export type AsrBatchControlAction = "pause" | "resume" | "cancel"; + +export type CreateAsrBatchJobInput = { + scope?: unknown; + rangeStart?: unknown; + rangeEnd?: unknown; + limit?: unknown; + awemeIds?: unknown; +}; + +export type AsrBatchJobItemSnapshot = { + index: number; + awemeId: string; + desc: string; + status: AsrBatchItemStatus; + tags: string[]; + error: string | null; + startedAt: string | null; + finishedAt: string | null; + durationMs: number | null; +}; + +export type AsrBatchJobSnapshot = { + id: string; + status: AsrBatchJobStatus; + scope: AsrBatchScope; + selection: { + sourceCount: number; + matchedCount: number; + selectedCount: number; + rangeStart: number; + rangeEnd: number; + requestedIdsCount: number; + }; + progress: { + total: number; + completed: number; + failed: number; + canceled: number; + pending: number; + running: number; + processed: number; + percent: number; + }; + currentItem: AsrBatchJobItemSnapshot | null; + pauseRequested: boolean; + cancelRequested: boolean; + error: string | null; + logs: { at: string; message: string; awemeId: string | null }[]; + createdAt: string; + startedAt: string | null; + updatedAt: string; + finishedAt: string | null; + items?: AsrBatchJobItemSnapshot[]; +}; + +type CandidateVideo = { + awemeId: string; + desc: string; + createdAt: string; + hasTranscript: boolean; + contentTags: string[]; +}; + +type InternalAsrBatchJob = Omit & { + items: AsrBatchJobItemSnapshot[]; + runner?: Promise; +}; + +type AsrBatchStore = { + jobs: Map; +}; + +const MAX_SOURCE_VIDEOS = 5000; +const MAX_JOB_ITEMS = 1000; +const MAX_JOB_HISTORY = 20; +const MAX_LOGS = 160; + +const globalForAsrBatchJobs = globalThis as typeof globalThis & { + __asrBatchJobs?: AsrBatchStore; +}; + +if (!globalForAsrBatchJobs.__asrBatchJobs) { + globalForAsrBatchJobs.__asrBatchJobs = { jobs: new Map() }; +} + +const store = globalForAsrBatchJobs.__asrBatchJobs; + +export class AsrBatchJobError extends Error { + status: number; + + constructor(message: string, status = 400) { + super(message); + this.name = "AsrBatchJobError"; + this.status = status; + } +} + +export function isTerminalAsrBatchStatus(status: AsrBatchJobStatus) { + return status === "completed" || status === "canceled" || status === "failed"; +} + +function nowIso() { + return new Date().toISOString(); +} + +function normalizeScope(scope: unknown): AsrBatchScope { + return typeof scope === "string" && + ASR_BATCH_SCOPES.includes(scope as AsrBatchScope) + ? (scope as AsrBatchScope) + : "missingTags"; +} + +function normalizePositiveInt(value: unknown) { + const numberValue = + typeof value === "number" + ? value + : typeof value === "string" && value.trim().length > 0 + ? Number.parseInt(value, 10) + : Number.NaN; + + return Number.isFinite(numberValue) && numberValue > 0 + ? Math.floor(numberValue) + : null; +} + +function normalizeAwemeIds(value: unknown) { + if (!Array.isArray(value)) return []; + + return [...new Set(value)] + .filter( + (id): id is string => typeof id === "string" && id.trim().length > 0, + ) + .map((id) => id.trim()) + .slice(0, MAX_JOB_ITEMS); +} + +function appendJobLog( + job: InternalAsrBatchJob, + message: string, + awemeId: string | null = null, +) { + job.logs.push({ at: nowIso(), message, awemeId }); + if (job.logs.length > MAX_LOGS) { + job.logs.splice(0, job.logs.length - MAX_LOGS); + } + job.updatedAt = nowIso(); +} + +function calculateProgress(job: InternalAsrBatchJob) { + const total = job.items.length; + const completed = job.items.filter( + (item) => item.status === "success", + ).length; + const failed = job.items.filter((item) => item.status === "failed").length; + const canceled = job.items.filter( + (item) => item.status === "canceled", + ).length; + const running = job.items.filter((item) => item.status === "running").length; + const pending = job.items.filter((item) => item.status === "pending").length; + const processed = completed + failed + canceled; + + return { + total, + completed, + failed, + canceled, + pending, + running, + processed, + percent: total > 0 ? Math.round((processed / total) * 100) : 0, + }; +} + +function serializeJob( + job: InternalAsrBatchJob, + includeItems = false, +): AsrBatchJobSnapshot { + const currentItem = + job.items.find((item) => item.status === "running") ?? null; + const snapshot: AsrBatchJobSnapshot = { + id: job.id, + status: job.status, + scope: job.scope, + selection: job.selection, + progress: calculateProgress(job), + currentItem, + pauseRequested: job.pauseRequested, + cancelRequested: job.cancelRequested, + error: job.error, + logs: job.logs, + createdAt: job.createdAt, + startedAt: job.startedAt, + updatedAt: job.updatedAt, + finishedAt: job.finishedAt, + }; + + if (includeItems) snapshot.items = job.items; + return snapshot; +} + +function trimJobHistory() { + const jobs = [...store.jobs.values()].sort( + (left, right) => + new Date(right.createdAt).getTime() - new Date(left.createdAt).getTime(), + ); + + for (const job of jobs.slice(MAX_JOB_HISTORY)) { + if (isTerminalAsrBatchStatus(job.status)) store.jobs.delete(job.id); + } +} + +function getActiveJob() { + return [...store.jobs.values()].find( + (job) => !isTerminalAsrBatchStatus(job.status), + ); +} + +async function readCandidateVideos(awemeIds: string[]) { + const videos = await prisma.video.findMany({ + ...(awemeIds.length > 0 ? { where: { aweme_id: { in: awemeIds } } } : {}), + orderBy: { created_at: "desc" }, + take: awemeIds.length > 0 ? awemeIds.length : MAX_SOURCE_VIDEOS, + select: { + aweme_id: true, + desc: true, + created_at: true, + manual_content_tags: true, + transcript: { + select: { + id: true, + content_tags: true, + }, + }, + }, + }); + + const orderMap = new Map(awemeIds.map((awemeId, index) => [awemeId, index])); + const sortedVideos = + awemeIds.length > 0 + ? [...videos].sort( + (left, right) => + (orderMap.get(left.aweme_id) ?? 0) - + (orderMap.get(right.aweme_id) ?? 0), + ) + : videos; + + return sortedVideos.map((video) => ({ + awemeId: video.aweme_id, + desc: video.desc, + createdAt: video.created_at.toISOString(), + hasTranscript: Boolean(video.transcript?.id), + contentTags: resolveEffectiveContentTags( + video.manual_content_tags, + video.transcript?.content_tags, + ), + })); +} + +function filterCandidatesByScope( + candidates: CandidateVideo[], + scope: AsrBatchScope, +) { + if (scope === "selected" || scope === "all") return candidates; + + return candidates.filter((candidate) => { + const hasTags = candidate.contentTags.length > 0; + if (scope === "missingTags") return !hasTags; + if (scope === "pending") return !candidate.hasTranscript; + if (scope === "transcribed") return candidate.hasTranscript; + if (scope === "tagged") return hasTags; + return true; + }); +} + +function resolveRange(matchedCount: number, input: CreateAsrBatchJobInput) { + const rangeStart = normalizePositiveInt(input.rangeStart) ?? 1; + const explicitRangeEnd = normalizePositiveInt(input.rangeEnd); + const limit = normalizePositiveInt(input.limit); + const requestedRangeEnd = + explicitRangeEnd ?? (limit ? rangeStart + limit - 1 : matchedCount); + const rangeEnd = Math.min( + Math.max(requestedRangeEnd, rangeStart), + matchedCount, + ); + + return { rangeStart, rangeEnd }; +} + +async function selectJobItems(input: CreateAsrBatchJobInput) { + const awemeIds = normalizeAwemeIds(input.awemeIds); + const scope = awemeIds.length > 0 ? "selected" : normalizeScope(input.scope); + const sourceCandidates = await readCandidateVideos(awemeIds); + const matchedCandidates = filterCandidatesByScope(sourceCandidates, scope); + const { rangeStart, rangeEnd } = resolveRange( + matchedCandidates.length, + input, + ); + const selectedCandidates = matchedCandidates + .slice(rangeStart - 1, rangeEnd) + .slice(0, MAX_JOB_ITEMS); + + return { + scope, + rangeStart, + rangeEnd, + sourceCount: sourceCandidates.length, + matchedCount: matchedCandidates.length, + requestedIdsCount: awemeIds.length, + selectedCandidates, + }; +} + +function launchRunner(job: InternalAsrBatchJob) { + if (job.runner) return; + + const runner = runAsrBatchJob(job.id).finally(() => { + const latestJob = store.jobs.get(job.id); + if (latestJob?.runner === runner) latestJob.runner = undefined; + }); + job.runner = runner; +} + +async function runAsrBatchJob(jobId: string) { + const job = store.jobs.get(jobId); + if (!job) return; + + job.status = "running"; + job.startedAt ??= nowIso(); + job.updatedAt = nowIso(); + appendJobLog(job, "任务开始执行"); + + try { + for (;;) { + if (job.cancelRequested) { + finalizeCanceledJob(job, "任务已取消"); + return; + } + + if (job.pauseRequested) { + job.status = "paused"; + job.updatedAt = nowIso(); + appendJobLog(job, "任务已暂停"); + return; + } + + const nextItem = job.items.find((item) => item.status === "pending"); + if (!nextItem) { + job.status = "completed"; + job.finishedAt = nowIso(); + job.updatedAt = nowIso(); + appendJobLog(job, "任务已完成"); + return; + } + + await runJobItem(job, nextItem); + } + } catch (error) { + job.status = "failed"; + job.error = error instanceof Error ? error.message : "任务执行失败"; + job.finishedAt = nowIso(); + job.updatedAt = nowIso(); + appendJobLog(job, job.error); + } +} + +async function runJobItem( + job: InternalAsrBatchJob, + item: AsrBatchJobItemSnapshot, +) { + const startedAt = Date.now(); + item.status = "running"; + item.startedAt = nowIso(); + item.error = null; + job.updatedAt = nowIso(); + appendJobLog( + job, + `开始处理第 ${item.index}/${job.items.length} 个视频`, + item.awemeId, + ); + + try { + const result = await transcriptAweme(item.awemeId, { + log: (message) => appendJobLog(job, message, item.awemeId), + }); + item.status = "success"; + item.tags = result.content_tags; + appendJobLog( + job, + `处理完成: ${result.content_tags.join("、")}`, + item.awemeId, + ); + } catch (error) { + item.status = "failed"; + item.error = error instanceof Error ? error.message : "处理失败"; + appendJobLog(job, `处理失败: ${item.error}`, item.awemeId); + } finally { + item.finishedAt = nowIso(); + item.durationMs = Date.now() - startedAt; + job.updatedAt = nowIso(); + } +} + +function finalizeCanceledJob(job: InternalAsrBatchJob, message: string) { + for (const item of job.items) { + if (item.status === "pending") { + item.status = "canceled"; + item.finishedAt = nowIso(); + } + } + job.status = "canceled"; + job.finishedAt = nowIso(); + job.updatedAt = nowIso(); + appendJobLog(job, message); +} + +export async function createAsrBatchJob(input: CreateAsrBatchJobInput) { + trimJobHistory(); + + const activeJob = getActiveJob(); + if (activeJob) { + throw new AsrBatchJobError("已有批量任务正在执行或暂停中", 409); + } + + const selection = await selectJobItems(input); + if (selection.selectedCandidates.length === 0) { + throw new AsrBatchJobError("没有匹配的视频可处理", 400); + } + + const createdAt = nowIso(); + const job: InternalAsrBatchJob = { + id: randomUUID(), + status: "queued", + scope: selection.scope, + selection: { + sourceCount: selection.sourceCount, + matchedCount: selection.matchedCount, + selectedCount: selection.selectedCandidates.length, + rangeStart: selection.rangeStart, + rangeEnd: selection.rangeEnd, + requestedIdsCount: selection.requestedIdsCount, + }, + currentItem: null, + pauseRequested: false, + cancelRequested: false, + error: null, + logs: [], + createdAt, + startedAt: null, + updatedAt: createdAt, + finishedAt: null, + items: selection.selectedCandidates.map((candidate, index) => ({ + index: index + 1, + awemeId: candidate.awemeId, + desc: candidate.desc, + status: "pending", + tags: [], + error: null, + startedAt: null, + finishedAt: null, + durationMs: null, + })), + }; + + store.jobs.set(job.id, job); + appendJobLog( + job, + `任务已创建，匹配 ${selection.matchedCount} 个，选中 ${selection.selectedCandidates.length} 个`, + ); + launchRunner(job); + + return serializeJob(job, true); +} + +export function listAsrBatchJobs() { + return [...store.jobs.values()] + .sort( + (left, right) => + new Date(right.createdAt).getTime() - + new Date(left.createdAt).getTime(), + ) + .map((job) => serializeJob(job)); +} + +export function getActiveAsrBatchJob() { + const activeJob = getActiveJob(); + return activeJob ? serializeJob(activeJob) : null; +} + +export function getAsrBatchJob(jobId: string, includeItems = true) { + const job = store.jobs.get(jobId); + return job ? serializeJob(job, includeItems) : null; +} + +export function controlAsrBatchJob(jobId: string, action: unknown) { + const job = store.jobs.get(jobId); + if (!job) throw new AsrBatchJobError("任务不存在", 404); + + if (action === "pause") { + if (isTerminalAsrBatchStatus(job.status)) return serializeJob(job, true); + job.pauseRequested = true; + job.cancelRequested = false; + if (!job.runner && job.status !== "paused") job.status = "paused"; + appendJobLog(job, "已请求暂停，当前视频处理完后暂停"); + return serializeJob(job, true); + } + + if (action === "resume" || action === "continue") { + if (isTerminalAsrBatchStatus(job.status)) { + throw new AsrBatchJobError("已结束的任务不能继续", 409); + } + if (job.cancelRequested) { + throw new AsrBatchJobError("已请求取消的任务不能继续", 409); + } + job.pauseRequested = false; + job.status = "running"; + appendJobLog(job, "任务继续执行"); + launchRunner(job); + return serializeJob(job, true); + } + + if (action === "cancel") { + if (isTerminalAsrBatchStatus(job.status)) return serializeJob(job, true); + job.cancelRequested = true; + job.pauseRequested = false; + if (!job.runner || job.status === "paused" || job.status === "queued") { + finalizeCanceledJob(job, "任务已取消"); + } else { + appendJobLog(job, "已请求取消，当前视频处理完后停止"); + } + return serializeJob(job, true); + } + + throw new AsrBatchJobError("不支持的任务操作", 400); +}
+ + 视频
+ toggleVideoSelection(video.aweme_id)} + className="h-4 w-4 rounded border-gray-300 text-blue-600" + aria-label="选择视频" + /> +
- {video.transcript ? ( - - ) : ( - - )} + "生成标签" + ) + ) : ( + "开始识别" + )} +