diff --git a/app/api/search-parallel/route.ts b/app/api/search-parallel/route.ts new file mode 100644 index 0000000..e312b6a --- /dev/null +++ b/app/api/search-parallel/route.ts @@ -0,0 +1,164 @@ +/** + * Parallel Streaming Search API Route + * Searches all sources in parallel and streams results immediately as they arrive + * No waiting - results flow in real-time + */ + +import { NextRequest } from 'next/server'; +import { searchVideos } from '@/lib/api/client'; +import { getSourceById } from '@/lib/api/video-sources'; + +export async function POST(request: NextRequest) { + const encoder = new TextEncoder(); + + const stream = new ReadableStream({ + async start(controller) { + try { + const body = await request.json(); + const { query, sources: sourceIds, page = 1 } = body; + + // Validate input + if (!query || typeof query !== 'string' || query.trim().length === 0) { + controller.enqueue(encoder.encode(`data: ${JSON.stringify({ + type: 'error', + message: 'Invalid query' + })}\n\n`)); + controller.close(); + return; + } + + // Get source configurations + const sources = sourceIds + .map((id: string) => getSourceById(id)) + .filter((source: any): source is NonNullable => source !== undefined); + + if (sources.length === 0) { + controller.enqueue(encoder.encode(`data: ${JSON.stringify({ + type: 'error', + message: 'No valid sources' + })}\n\n`)); + controller.close(); + return; + } + + // Send initial status + controller.enqueue(encoder.encode(`data: ${JSON.stringify({ + type: 'start', + totalSources: sources.length + })}\n\n`)); + + console.log(`[Search Parallel] Starting search for "${query}" across ${sources.length} sources`); + + // Track progress + let completedSources = 0; + let totalVideosFound = 0; + + // Search all sources in PARALLEL - don't wait for all to finish + const searchPromises = sources.map(async (source: any) => { + try { + console.log(`[Search Parallel] Searching source: ${source.id} (${getSourceDisplayName(source.id)})`); + + // Search this source + const result = await searchVideos(query.trim(), [source], page); + const videos = result[0]?.results || []; + + completedSources++; + totalVideosFound += videos.length; + + console.log(`[Search Parallel] Source ${source.id} completed: ${videos.length} videos found`); + + // Stream videos immediately as they arrive + if (videos.length > 0) { + controller.enqueue(encoder.encode(`data: ${JSON.stringify({ + type: 'videos', + videos: videos.map((video: any) => ({ + ...video, + isVerifying: true, // Mark for verification + sourceDisplayName: getSourceDisplayName(source.id), + })), + source: source.id, + completedSources, + totalSources: sources.length + })}\n\n`)); + } + + // Send progress update + controller.enqueue(encoder.encode(`data: ${JSON.stringify({ + type: 'progress', + completedSources, + totalSources: sources.length, + totalVideosFound + })}\n\n`)); + + } catch (error) { + // Log error but continue with other sources + console.error(`[Search Parallel] Source ${source.id} failed:`, error); + completedSources++; + + controller.enqueue(encoder.encode(`data: ${JSON.stringify({ + type: 'progress', + completedSources, + totalSources: sources.length, + totalVideosFound + })}\n\n`)); + } + }); + + // Wait for all sources to complete + await Promise.all(searchPromises); + + console.log(`[Search Parallel] Search complete: ${totalVideosFound} total videos found from ${completedSources}/${sources.length} sources`); + + // Send completion signal + controller.enqueue(encoder.encode(`data: ${JSON.stringify({ + type: 'complete', + totalVideosFound, + totalSources: sources.length + })}\n\n`)); + + controller.close(); + + } catch (error) { + console.error('Search error:', error); + controller.enqueue(encoder.encode(`data: ${JSON.stringify({ + type: 'error', + message: error instanceof Error ? error.message : 'Unknown error' + })}\n\n`)); + controller.close(); + } + } + }); + + return new Response(stream, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + 'Connection': 'keep-alive', + }, + }); +} + +/** + * Get display name for source + */ +function getSourceDisplayName(sourceId: string): string { + const sourceNames: Record = { + 'dytt': '电影天堂', + 'ruyi': '如意', + 'baofeng': '暴风', + 'tianya': '天涯', + 'feifan': '非凡影视', + 'sanliuling': '360', + 'wolong': '卧龙', + 'jisu': '极速', + 'mozhua': '魔爪', + 'modu': '魔都', + 'zuida': '最大', + 'yinghua': '樱花', + 'baiduyun': '百度云', + 'wujin': '无尽', + 'wangwang': '旺旺', + 'ikun': 'iKun', + }; + return sourceNames[sourceId] || sourceId; +} diff --git a/app/api/search/route.ts b/app/api/search/route.ts index 17ca6c1..10ef6d8 100644 --- a/app/api/search/route.ts +++ b/app/api/search/route.ts @@ -48,22 +48,22 @@ export async function POST(request: NextRequest) { // Get source name mapping const getSourceName = (sourceId: string): string => { const sourceNames: Record = { - 'custom_0': '电影天堂', - 'custom_1': '如意', - 'custom_2': '暴风', - 'custom_3': '天涯', - 'custom_4': '非凡影视', - 'custom_5': '360', - 'custom_6': '卧龙', - 'custom_7': '极速', - 'custom_8': '魔爪', - 'custom_9': '魔都', - 'custom_10': '海外看', - 'custom_11': '新浪', - 'custom_12': '光速', - 'custom_13': '红牛', - 'custom_14': '樱花', - 'custom_15': '飞速', + 'dytt': '电影天堂', + 'ruyi': '如意', + 'baofeng': '暴风', + 'tianya': '天涯', + 'feifan': '非凡影视', + 'sanliuling': '360', + 'wolong': '卧龙', + 'jisu': '极速', + 'mozhua': '魔爪', + 'modu': '魔都', + 'zuida': '最大', + 'yinghua': '樱花', + 'baiduyun': '百度云', + 'wujin': '无尽', + 'wangwang': '旺旺', + 'ikun': 'iKun', }; return sourceNames[sourceId] || sourceId; }; @@ -160,22 +160,22 @@ export async function GET(request: NextRequest) { // Get source name mapping const getSourceName = (sourceId: string): string => { const sourceNames: Record = { - 'custom_0': '电影天堂', - 'custom_1': '如意', - 'custom_2': '暴风', - 'custom_3': '天涯', - 'custom_4': '非凡影视', - 'custom_5': '360', - 'custom_6': '卧龙', - 'custom_7': '极速', - 'custom_8': '魔爪', - 'custom_9': '魔都', - 'custom_10': '海外看', - 'custom_11': '新浪', - 'custom_12': '光速', - 'custom_13': '红牛', - 'custom_14': '樱花', - 'custom_15': '飞速', + 'dytt': '电影天堂', + 'ruyi': '如意', + 'baofeng': '暴风', + 'tianya': '天涯', + 'feifan': '非凡影视', + 'sanliuling': '360', + 'wolong': '卧龙', + 'jisu': '极速', + 'mozhua': '魔爪', + 'modu': '魔都', + 'zuida': '最大', + 'yinghua': '樱花', + 'baiduyun': '百度云', + 'wujin': '无尽', + 'wangwang': '旺旺', + 'ikun': 'iKun', }; return sourceNames[sourceId] || sourceId; }; diff --git a/app/page.tsx b/app/page.tsx index c8bf2a5..686269e 100644 --- a/app/page.tsx +++ b/app/page.tsx @@ -11,7 +11,7 @@ import { EmptyState } from '@/components/search/EmptyState'; import { NoResults } from '@/components/search/NoResults'; import { ResultsHeader } from '@/components/search/ResultsHeader'; import { useSearchCache } from '@/lib/hooks/useSearchCache'; -import { useSearchStream } from '@/lib/hooks/useSearchStream'; +import { useParallelSearch } from '@/lib/hooks/useParallelSearch'; function HomePage() { const router = useRouter(); @@ -27,16 +27,15 @@ function HomePage() { loading, results, availableSources, - checkedSources, - searchStage, - checkedVideos, - totalVideos, - currentSource, + completedSources, + totalSources, + totalVideosFound, performSearch, resetSearch, - } = useSearchStream( + loadCachedResults, + } = useParallelSearch( saveToCache, - (q) => router.replace(`/?q=${encodeURIComponent(q)}`, { scroll: false }) + (q: string) => router.replace(`/?q=${encodeURIComponent(q)}`, { scroll: false }) ); // Load cached results on mount @@ -49,21 +48,21 @@ function HomePage() { if (urlQuery) { setQuery(urlQuery); - if (cached && cached.query === urlQuery) { - console.log('📦 Loading cached results for:', urlQuery); - // Note: Would need to set results here if we expose setState from hook + if (cached && cached.query === urlQuery && cached.results.length > 0) { + console.log('📦 Loading cached results for:', urlQuery, cached.results.length, 'videos'); setHasSearched(true); + loadCachedResults(cached.results, cached.availableSources); } else { console.log('🔍 Auto-searching for URL query:', urlQuery); setTimeout(() => handleSearch(urlQuery), 100); } } - }, [searchParams]); + }, [searchParams, loadFromCache, loadCachedResults]); const handleSearch = (searchQuery: string) => { setQuery(searchQuery); setHasSearched(true); - performSearch(searchQuery, true); + performSearch(searchQuery); }; const handleReset = () => { @@ -118,12 +117,12 @@ function HomePage() { onSearch={handleSearch} isLoading={loading} initialQuery={query} - currentSource={currentSource} - checkedSources={checkedSources} - totalSources={16} - checkedVideos={checkedVideos} - totalVideos={totalVideos} - searchStage={searchStage} + currentSource="" + checkedSources={completedSources} + totalSources={totalSources} + checkedVideos={0} + totalVideos={totalVideosFound} + searchStage="searching" /> @@ -133,8 +132,8 @@ function HomePage() { diff --git a/components/search/VideoGrid.tsx b/components/search/VideoGrid.tsx index e6f69c5..e52b5cc 100644 --- a/components/search/VideoGrid.tsx +++ b/components/search/VideoGrid.tsx @@ -15,6 +15,7 @@ interface Video { source: string; sourceName?: string; isNew?: boolean; + isVerifying?: boolean; } interface VideoGridProps { @@ -68,6 +69,15 @@ export function VideoGrid({ videos, className = '' }: VideoGridProps) { )} + {/* Verifying Badge - Top Right */} + {video.isVerifying && ( +
+ + 验证中 + +
+ )} + {/* Overlay */}
diff --git a/lib/hooks/useParallelSearch.ts b/lib/hooks/useParallelSearch.ts new file mode 100644 index 0000000..776c137 --- /dev/null +++ b/lib/hooks/useParallelSearch.ts @@ -0,0 +1,313 @@ +'use client'; + +import { useState, useRef, useCallback } from 'react'; +import { getSourceName, SOURCE_IDS } from '@/lib/utils/source-names'; +import { checkVideoAvailability } from '@/lib/utils/source-checker'; + +interface Video { + vod_id: string; + vod_name: string; + vod_pic?: string; + vod_remarks?: string; + vod_year?: string; + type_name?: string; + source: string; + sourceName?: string; + isNew?: boolean; + isVerifying?: boolean; + vod_play_url?: string; +} + +export interface ParallelSearchResult { + loading: boolean; + results: Video[]; + availableSources: any[]; + completedSources: number; + totalSources: number; + totalVideosFound: number; + performSearch: (query: string) => Promise; + resetSearch: () => void; + loadCachedResults: (results: Video[], sources: any[]) => void; +} + +export function useParallelSearch( + onCacheUpdate: (query: string, results: any[], sources: any[]) => void, + onUrlUpdate: (query: string) => void +): ParallelSearchResult { + const [loading, setLoading] = useState(false); + const [results, setResults] = useState([]); + const [availableSources, setAvailableSources] = useState([]); + const [completedSources, setCompletedSources] = useState(0); + const [totalSources, setTotalSources] = useState(0); + const [totalVideosFound, setTotalVideosFound] = useState(0); + + const abortControllerRef = useRef(null); + const verificationQueueRef = useRef([]); + const verifyingCountRef = useRef(0); + const allVideosReceivedRef = useRef(false); + const MAX_CONCURRENT_VERIFICATIONS = 15; + + /** + * Process verification queue + * Verifies videos in parallel with a max concurrency limit + */ + const processVerificationQueue = useCallback(async () => { + while (verificationQueueRef.current.length > 0 && verifyingCountRef.current < MAX_CONCURRENT_VERIFICATIONS) { + const video = verificationQueueRef.current.shift(); + if (!video) continue; + + verifyingCountRef.current++; + + // Verify in background + checkVideoAvailability(video) + .then((isValid) => { + verifyingCountRef.current--; + + setResults((prev) => { + if (isValid) { + // Remove verifying badge + return prev.map((v) => + v.vod_id === video.vod_id && v.source === video.source + ? { ...v, isVerifying: false } + : v + ); + } else { + // Remove invalid video + const removedVideo = prev.find( + (v) => v.vod_id === video.vod_id && v.source === video.source + ); + + if (removedVideo && allVideosReceivedRef.current) { + // Update source count when removing video after search is complete + setAvailableSources((sources) => + sources.map((s) => + s.id === removedVideo.source + ? { ...s, count: Math.max(0, s.count - 1) } + : s + ).filter(s => s.count > 0) + ); + } + + return prev.filter( + (v) => !(v.vod_id === video.vod_id && v.source === video.source) + ); + } + }); + + // Continue processing queue + processVerificationQueue(); + }) + .catch((error) => { + console.error('Verification error:', error); + verifyingCountRef.current--; + + // Remove video on error + setResults((prev) => { + const removedVideo = prev.find( + (v) => v.vod_id === video.vod_id && v.source === video.source + ); + + if (removedVideo && allVideosReceivedRef.current) { + // Update source count when removing video + setAvailableSources((sources) => + sources.map((s) => + s.id === removedVideo.source + ? { ...s, count: Math.max(0, s.count - 1) } + : s + ).filter(s => s.count > 0) + ); + } + + return prev.filter( + (v) => !(v.vod_id === video.vod_id && v.source === video.source) + ); + }); + + // Continue processing queue + processVerificationQueue(); + }); + } + }, []); + + /** + * Add videos to verification queue + */ + const queueVideosForVerification = useCallback((videos: Video[]) => { + verificationQueueRef.current.push(...videos); + processVerificationQueue(); + }, [processVerificationQueue]); + + /** + * Perform parallel search with streaming results + */ + const performSearch = useCallback(async (searchQuery: string) => { + if (!searchQuery.trim() || loading) return; + + // Abort any ongoing search + if (abortControllerRef.current) { + abortControllerRef.current.abort(); + } + abortControllerRef.current = new AbortController(); + + // Reset state + setLoading(true); + setResults([]); + setAvailableSources([]); + setCompletedSources(0); + setTotalSources(0); + setTotalVideosFound(0); + verificationQueueRef.current = []; + verifyingCountRef.current = 0; + allVideosReceivedRef.current = false; + + // Update URL + onUrlUpdate(searchQuery); + + try { + const response = await fetch('/api/search-parallel', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ query: searchQuery, sources: SOURCE_IDS }), + signal: abortControllerRef.current.signal, + }); + + if (!response.ok) throw new Error('Search failed'); + + const reader = response.body?.getReader(); + const decoder = new TextDecoder(); + if (!reader) throw new Error('No response stream'); + + let buffer = ''; + const sourcesMap = new Map(); + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines.pop() || ''; + + for (const line of lines) { + if (!line.startsWith('data: ')) continue; + + try { + const data = JSON.parse(line.slice(6)); + + if (data.type === 'start') { + setTotalSources(data.totalSources); + console.log(`[useParallelSearch] Started search for ${data.totalSources} sources`); + } + else if (data.type === 'videos') { + const newVideos: Video[] = data.videos.map((video: any) => ({ + ...video, + sourceName: video.sourceDisplayName || getSourceName(video.source), + isNew: true, + isVerifying: true, // All videos start as verifying + })); + + console.log(`[useParallelSearch] Received ${newVideos.length} videos from source ${data.source}`); + + // Add videos to display immediately + setResults((prev) => [...prev, ...newVideos]); + + // Queue videos for verification + queueVideosForVerification(newVideos); + + // Update source stats + if (!sourcesMap.has(data.source)) { + sourcesMap.set(data.source, { + count: newVideos.length, + name: newVideos[0]?.sourceName || data.source, + }); + } + } + else if (data.type === 'progress') { + setCompletedSources(data.completedSources); + setTotalVideosFound(data.totalVideosFound); + } + else if (data.type === 'complete') { + setLoading(false); + allVideosReceivedRef.current = true; + + console.log(`[useParallelSearch] Search complete: ${data.totalVideosFound} videos found`); + console.log(`[useParallelSearch] Sources with videos: ${sourcesMap.size}`); + + // Update available sources with correct property names + const sources = Array.from(sourcesMap.entries()).map(([id, info]) => ({ + id: id, // Changed from sourceId to id + name: info.name, // Changed from sourceName to name + count: info.count, + })); + setAvailableSources(sources); + + console.log('[useParallelSearch] Available sources:', sources); + + // Cache results - wait a bit for current results to be in state + setTimeout(() => { + setResults((currentResults) => { + onCacheUpdate(searchQuery, currentResults, sources); + return currentResults; + }); + }, 100); + } + else if (data.type === 'error') { + console.error('Search error:', data.message); + setLoading(false); + } + } catch (error) { + console.error('Error parsing stream data:', error); + } + } + } + } catch (error) { + if (error instanceof Error && error.name === 'AbortError') { + console.log('Search aborted'); + } else { + console.error('Search error:', error); + } + setLoading(false); + } + }, [loading, onUrlUpdate, onCacheUpdate, queueVideosForVerification]); + + /** + * Reset search state + */ + const resetSearch = useCallback(() => { + if (abortControllerRef.current) { + abortControllerRef.current.abort(); + } + setLoading(false); + setResults([]); + setAvailableSources([]); + setCompletedSources(0); + setTotalSources(0); + setTotalVideosFound(0); + verificationQueueRef.current = []; + verifyingCountRef.current = 0; + allVideosReceivedRef.current = false; + }, []); + + /** + * Load cached results + */ + const loadCachedResults = useCallback((cachedResults: Video[], cachedSources: any[]) => { + console.log('[useParallelSearch] Loading cached results:', cachedResults.length, 'videos'); + setResults(cachedResults); + setAvailableSources(cachedSources); + setTotalVideosFound(cachedResults.length); + allVideosReceivedRef.current = true; + }, []); + + return { + loading, + results, + availableSources, + completedSources, + totalSources, + totalVideosFound, + performSearch, + resetSearch, + loadCachedResults, + }; +} diff --git a/lib/utils/source-names.ts b/lib/utils/source-names.ts index 612c045..73eeb2e 100644 --- a/lib/utils/source-names.ts +++ b/lib/utils/source-names.ts @@ -21,7 +21,7 @@ export function getSourceName(sourceId: string): string { } export const SOURCE_IDS = [ - 'dytt', 'ruyi', 'baofeng', 'tianya', 'feifan', - 'sanliuling', 'wolong', 'jisu', 'mozhua', 'modu', - 'zuida', 'yinghua', 'baiduyun', 'wujin', 'wangwang', 'ikun' + 'dytt', 'ruyi', 'baofeng', 'tianya', 'feifan', 'sanliuling', + 'wolong', 'jisu', 'mozhua', 'modu', 'zuida', 'yinghua', + 'baiduyun', 'wujin', 'wangwang', 'ikun' ];