admin/momentry_studio
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Revert "perf: lazy load face group thumbnails + increase concurrency to 32"

Browse Source 
This reverts commit 26b94f7243.
This commit is contained in:
Momentry Studio
2026-08-01 01:58:01 +08:00
parent 26b94f7243
commit 5be57aec7d
2 changed files with 248 additions and 317 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
src/store.ts
View File

@@ -21,7 +21,6 @@ const UNASSIGNED_PER_PAGE = 20
export const thumbnailsCache = ref<Record<string, string>>({})
export const profilesCache = ref<Record<string, string>>({})
export const faceThumbsCache = ref<Record<string, string>>({})
export const frameThumbsCache = ref<Record<string, string>>({})
export const poseCache = ref<Record<string, PoseResponse>>({})
export const appearanceCache = ref<Record<string, AppearanceResponse>>({})
@@ -124,11 +123,20 @@ export interface TkgStats {
total_edges: number
}
export interface IdentityAgentStats {
clusters: number
identities_created: number
tmdb_matches: number
speaker_bindings: number
confirmations: number
}
export interface FileStats {
json: JsonStats
postgresql: PostgresqlStats
qdrant: QdrantStats
tkg: TkgStats
identity_agent: IdentityAgentStats
}
export const processorCountsCache = ref<Record<string, Record<string, ProcessorOutputInfo>>>({})
@@ -324,7 +332,7 @@ loadProgressCacheFromStorage()
const thumbQueue: (() => Promise<void>)[] = []
let activeThumbLoads = 0
const MAX_CONCURRENT = 32
const MAX_CONCURRENT = 16
const profileQueue: (() => Promise<void>)[] = []
let activeProfileLoads = 0
@@ -333,7 +341,6 @@ const MAX_PROFILE_CONCURRENT = 4
const loadingThumbs = new Set<string>()
const loadingProfiles = new Set<string>()
const loadingFaceThumbs = new Set<string>()
const loadingFrameThumbs = new Set<string>()
const loadingProcessorCounts = new Set<string>()
const pollingProgress = new Set<string>()
export const pollingProgressSet = pollingProgress
@@ -369,10 +376,11 @@ export function stopStatsAutoRefresh() {
export function startProcessingFilesPolling() {
if (processingPollInterval) return
processingPollInterval = setInterval(async () => {
const scanningFiles = filesCache.value.filter((f: any) => f.status === 'scanning' && f.file_uuid)
if (scanningFiles.length === 0) return
await Promise.all(scanningFiles.map((f: any) => syncFileStatus(f.file_uuid)))
}, 5000)
const processingFiles = filesCache.value.filter((f: any) => f.status === 'processing' && f.file_uuid)
if (processingFiles.length === 0) return
// Use Promise.all to sync in parallel, not sequentially
await Promise.all(processingFiles.map((f: any) => syncFileStatus(f.file_uuid)))
}, 10000) // Increased from 5s to 10s
}
export function stopProcessingFilesPolling() {
@@ -401,20 +409,16 @@ export function startFilesWatch() {
filesCache.value = newFiles
}
// Check status changes for existing files (never regress from agent_ready)
// Check status changes for existing files
for (const newFile of newFiles) {
if (!newFile.file_uuid) continue
const oldFile = filesCache.value.find((f: any) => f.file_uuid === newFile.file_uuid)
if (!oldFile || oldFile.status === newFile.status) continue
if (oldFile.status === 'agent_ready') continue
// files/scan often returns stale 'scanning' — resolve via syncFileStatus
if (newFile.status === 'scanning') {
syncFileStatus(newFile.file_uuid)
continue
}
const idx = filesCache.value.findIndex((f: any) => f.file_uuid === newFile.file_uuid)
if (idx >= 0) {
filesCache.value[idx] = newFile
if (oldFile && oldFile.status !== newFile.status) {
// Status changed, update cache
const idx = filesCache.value.findIndex((f: any) => f.file_uuid === newFile.file_uuid)
if (idx >= 0) {
filesCache.value[idx] = newFile
}
}
}
} catch (e) {
@@ -475,10 +479,10 @@ let _lastLoadedFileUuid: string | undefined = undefined
export async function ensureFiles() {
console.error('[ensureFiles] starting, filesLoaded.value:', filesLoaded.value)
if (filesLoaded.value && filesCache.value.length > 0) return
if (filesLoaded.value) return
if (_filesLoading) {
await new Promise<void>(r => { const check = setInterval(() => { if (filesLoaded.value && filesCache.value.length > 0 || !_filesLoading) { clearInterval(check); r() } }, 100) })
if (filesLoaded.value && filesCache.value.length > 0) return
await new Promise<void>(r => { const check = setInterval(() => { if (filesLoaded.value || !_filesLoading) { clearInterval(check); r() } }, 100) })
return
}
_filesLoading = true
try {
@@ -488,7 +492,7 @@ export async function ensureFiles() {
console.error('Failed to load files:', e)
} finally {
_filesLoading = false
filesLoaded.value = filesCache.value.length > 0
filesLoaded.value = true
}
}
@@ -582,7 +586,7 @@ export async function ensureUnassignedTraces(fileUuid: string, page?: number) {
}
export function loadTraceThumb(t: any) {
if (!t.file_uuid || t.trace_id == null) return
if (!t.file_uuid || !t.trace_id) return
const key = `${t.trace_id}-${t.file_uuid}-${t.best_face_id || 0}`
if (faceThumbsCache.value[key] || loadingFaceThumbs.has(key)) return
loadingFaceThumbs.add(key)
@@ -605,33 +609,6 @@ export function loadTraceThumb(t: any) {
})
}
export function frameThumbKey(t: any): string {
return `${t.trace_id}-${t.file_uuid}`
}
export function loadFrameThumb(t: any) {
if (!t.file_uuid || t.trace_id == null) return
const key = frameThumbKey(t)
if (frameThumbsCache.value[key] || loadingFrameThumbs.has(key)) return
loadingFrameThumbs.add(key)
queueThumb(async () => {
try {
const frame = t.best_face_frame ?? t.start_frame ?? 0
const url = `/api/v1/media/frame?file_uuid=${t.file_uuid}&frame=${frame}`
const resp = await fetch(url)
if (resp.ok) {
const blob = await resp.blob()
const dataUrl = URL.createObjectURL(blob)
frameThumbsCache.value[key] = dataUrl
}
} catch (e) {
console.error('loadFrameThumb failed:', key, e)
} finally {
loadingFrameThumbs.delete(key)
}
})
}
export function loadThumbnail(uuid: string, frame = 30) {
const key = `${uuid}:${frame}`
if (!uuid || thumbnailsCache.value[key] || loadingThumbs.has(key)) return
@@ -819,13 +796,17 @@ export async function pollProgress(fileUuid: string) {
gpu_percent: data.gpu_percent,
memory_percent: data.memory_percent,
}
fileStatusCache.value[fileUuid] = 'processing'
const f = filesCache.value.find((x: any) => x.file_uuid === fileUuid)
if (f) f.status = 'processing'
saveProgressCacheToStorage()
const allComplete = data.processors?.every((p: any) => p.status === 'complete' || p.status === 'failed')
if (allComplete || data.overall_progress >= 100) {
stopPolling(fileUuid)
fileStatusCache.value[fileUuid] = 'completed'
if (f) f.status = 'completed'
saveProgressCacheToStorage()
await loadProcessorCounts(fileUuid)
await syncFileStatus(fileUuid)
break
}
await new Promise(r => setTimeout(r, 2000))
@@ -856,7 +837,25 @@ export function clearFileStatusCache(fileUuid: string) {
export async function refreshFileStatus(fileUuid: string) {
try {
await loadPipelineStats(fileUuid, true)
const pipeline = await loadPipelineStats(fileUuid, true)
if (pipeline) {
const isComplete = pipeline.overall_progress >= 1.0 || pipeline.stages?.every((s: any) => s.status === 'completed' || s.status === 'failed')
const isRunning = pipeline.stages?.some((s: any) => s.status === 'running')
if (isComplete) {
fileStatusCache.value[fileUuid] = 'completed'
await loadFileStats(fileUuid, true)
} else if (isRunning) {
fileStatusCache.value[fileUuid] = 'processing'
}
// Also update the file object in filesCache
const f = filesCache.value.find((x: any) => x.file_uuid === fileUuid)
if (f) {
if (isComplete) f.status = 'completed'
else if (isRunning) f.status = 'processing'
}
}
} catch (e) {
console.error('refreshFileStatus failed:', fileUuid, e)
}
@@ -875,13 +874,20 @@ export async function syncFileStatus(fileUuid: string): Promise<any> {
if (result?.status) {
const newStatus = result.status
const oldStatus = fileStatusCache.value[fileUuid]
fileStatusCache.value[fileUuid] = newStatus
const f = filesCache.value.find((x: any) => x.file_uuid === fileUuid)
if (f) {
// If status changed from processing to completed, clear pipeline cache
if (oldStatus === 'processing' && newStatus === 'completed') {
delete pipelineStatsCache.value[fileUuid]
delete fileProgressCache.value[fileUuid]
}
f.status = newStatus
if (newStatus === 'scanning') {
if (newStatus === 'processing') {
await loadPipelineStats(fileUuid, true)
}
}
@@ -895,8 +901,8 @@ export async function syncFileStatus(fileUuid: string): Promise<any> {
}
export async function syncAllProcessingFiles(): Promise<void> {
const scanningFiles = filesCache.value.filter((f: any) => f.status === 'scanning' && f.file_uuid)
for (const f of scanningFiles) {
const processingFiles = filesCache.value.filter((f: any) => f.status === 'processing' && f.file_uuid)
for (const f of processingFiles) {
await syncFileStatus(f.file_uuid)
}
}
@@ -915,10 +921,11 @@ export async function loadPipelineStats(fileUuid: string, force = false): Promis
const WEIGHT_MAP: Record<string, number> = {
processors: 0.30,
rule1_ingestion: 0.05,
face_tracing: 0.08,
tkg_nodes: 0.23,
tkg_edges: 0.18,
rule2_ingestion: 0.16
face_tracing: 0.05,
identity_agent: 0.10,
tkg_nodes: 0.20,
tkg_edges: 0.15,
rule2_ingestion: 0.15
}
const mappedStages = stages.map((s: any) => ({
@@ -929,9 +936,7 @@ export async function loadPipelineStats(fileUuid: string, force = false): Promis
}))
// 用前端权重重新计算 overall_progress（不信任 API 返回的值）
const calculatedRaw = mappedStages.reduce((sum, s) => sum + s.weight * s.progress, 0)
const allDone = mappedStages.length > 0 && mappedStages.every((s: any) => s.status === 'completed' || s.status === 'failed')
const calculatedProgress = allDone ? 1.0 : Math.min(calculatedRaw, 1.0)
const calculatedProgress = mappedStages.reduce((sum, s) => sum + s.weight * s.progress, 0)
const pipelineStats: PipelineStats = {
overall_progress: calculatedProgress,
@@ -942,18 +947,31 @@ export async function loadPipelineStats(fileUuid: string, force = false): Promis
pipelineStatsCache.value = { ...pipelineStatsCache.value, [fileUuid]: pipelineStats }
// If pipeline is complete, update file status
if (calculatedProgress >= 1.0) {
const f = filesCache.value.find((x: any) => x.file_uuid === fileUuid)
if (f && f.status !== 'completed') {
f.status = 'completed'
fileStatusCache.value[fileUuid] = 'completed'
}
}
return pipelineStats
} catch (e) {
console.warn('get_pipeline_stats failed, falling back to get_progress:', fileUuid, e)
try {
const oldData: any = await apiCall('get_progress', { fileUuid })
// 检查所有处理器是否都完成
const allComplete = oldData.processors?.length > 0 && oldData.processors?.every((p: any) => p.status === 'complete' || p.status === 'failed')
const hasRunning = oldData.processors?.some((p: any) => p.status === 'running')
// 如果都完成，显示100%
const progress = allComplete ? 1.0 : (oldData.overall_progress || 0) / 100
const status = allComplete ? 'completed' : (hasRunning ? 'running' : 'pending')
console.log('[loadPipelineStats fallback]', fileUuid, { allComplete, hasRunning, progress, status, processorCount: oldData.processors?.length })
const fallbackStats: PipelineStats = {
overall_progress: progress,
stages: [
@@ -961,7 +979,6 @@ export async function loadPipelineStats(fileUuid: string, force = false): Promis
]
}
pipelineStatsCache.value = { ...pipelineStatsCache.value, [fileUuid]: fallbackStats }
return fallbackStats
} catch (fallbackError) {
console.error('Both pipeline APIs failed:', fileUuid, fallbackError)
@@ -1037,11 +1054,25 @@ export async function loadFileStats(fileUuid: string, force = false): Promise<Fi
},
total_nodes: data.tkg?.total_nodes || 0,
total_edges: data.tkg?.total_edges || 0
},
identity_agent: {
clusters: data.identity_agent?.clusters || 0,
identities_created: data.identity_agent?.identities_created || 0,
tmdb_matches: data.identity_agent?.tmdb_matches || 0,
speaker_bindings: data.identity_agent?.speaker_bindings || 0,
confirmations: data.identity_agent?.confirmations || 0
}
}
fileStatsCache.value = { ...fileStatsCache.value, [fileUuid]: fileStats }
if (data.status === 'completed') {
pipelineStatsCache.value[fileUuid] = {
overall_progress: 1.0,
stages: [{ name: 'processors', weight: 1.0, progress: 1.0, status: 'completed' }]
}
}
return fileStats
} catch (e) {
console.warn('get_file_stats failed, falling back to get_processor_counts:', fileUuid, e)
@@ -1059,7 +1090,8 @@ export async function loadFileStats(fileUuid: string, force = false): Promise<Fi
},
postgresql: { sentence_chunks: 0, trace_chunks: 0, relationship_chunks: 0, identities: 0, file_identities: 0 },
qdrant: { faces: 0, face_traces: 0, face_identities: 0, text_chunks: 0, speakers: 0 },
tkg: { nodes: { character: 0, object: 0, scene: 0, action: 0, location: 0, event: 0, dialogue: 0, emotion: 0, relation: 0 }, edges: { appear_in: 0, interact_with: 0, located_at: 0, perform_action: 0, cause_effect: 0, temporal_link: 0, speak_to: 0, feel: 0 }, total_nodes: 0, total_edges: 0 }
tkg: { nodes: { character: 0, object: 0, scene: 0, action: 0, location: 0, event: 0, dialogue: 0, emotion: 0, relation: 0 }, edges: { appear_in: 0, interact_with: 0, located_at: 0, perform_action: 0, cause_effect: 0, temporal_link: 0, speak_to: 0, feel: 0 }, total_nodes: 0, total_edges: 0 },
identity_agent: { clusters: 0, identities_created: 0, tmdb_matches: 0, speaker_bindings: 0, confirmations: 0 }
}
// 从processorCounts填充json.processors
@@ -1091,6 +1123,39 @@ export function getFileStats(fileUuid: string): FileStats | null {
return fileStatsCache.value[fileUuid] || null
}
export async function pollPipelineProgress(fileUuid: string) {
if (!fileUuid || pollingProgress.has(fileUuid)) return
pollingProgress.add(fileUuid)
try {
while (pollingProgress.has(fileUuid)) {
const stats = await loadPipelineStats(fileUuid, true)
if (stats) {
const idx = filesCache.value.findIndex((x: any) => x.file_uuid === fileUuid)
if (idx === -1) continue
fileStatusCache.value[fileUuid] = 'processing'
filesCache.value.splice(idx, 1, { ...filesCache.value[idx], status: 'processing' })
saveProgressCacheToStorage()
const allComplete = stats.stages.every(s => s.status === 'completed' || s.status === 'failed')
if (allComplete || stats.overall_progress >= 1.0) {
stopPolling(fileUuid)
fileStatusCache.value[fileUuid] = 'completed'
filesCache.value.splice(idx, 1, { ...filesCache.value[idx], status: 'completed' })
saveProgressCacheToStorage()
await loadFileStats(fileUuid, true)
break
}
}
await new Promise(r => setTimeout(r, 2000))
}
} catch (e) {
console.error('pollPipelineProgress failed:', fileUuid, e)
} finally {
pollingProgress.delete(fileUuid)
}
}
export function invalidateFiles() {
filesLoaded.value = false
filesCache.value = []