feat: Implement parallel search functionality and update source name mappings

2026-08-16 17:23:43 +08:00 · 2025-11-17 14:29:31 +08:00
parent ad9e75bc8c
commit 2802c694d0
6 changed files with 542 additions and 56 deletions
@@ -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<typeof source> => 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<string, string> = {
+    'dytt': '电影天堂',
+    'ruyi': '如意',
+    'baofeng': '暴风',
+    'tianya': '天涯',
+    'feifan': '非凡影视',
+    'sanliuling': '360',
+    'wolong': '卧龙',
+    'jisu': '极速',
+    'mozhua': '魔爪',
+    'modu': '魔都',
+    'zuida': '最大',
+    'yinghua': '樱花',
+    'baiduyun': '百度云',
+    'wujin': '无尽',
+    'wangwang': '旺旺',
+    'ikun': 'iKun',
+  };
+  return sourceNames[sourceId] || sourceId;
+}
@@ -48,22 +48,22 @@ export async function POST(request: NextRequest) {
    // Get source name mapping
    const getSourceName = (sourceId: string): string => {
      const sourceNames: Record<string, string> = {
-        '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<string, string> = {
-        '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;
    };
@@ -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"
          />
        </div>

@@ -133,8 +132,8 @@ function HomePage() {
            <ResultsHeader
              loading={loading}
              resultsCount={results.length}
-              checkedVideos={checkedVideos}
-              totalVideos={totalVideos}
+              checkedVideos={0}
+              totalVideos={totalVideosFound}
              availableSources={availableSources}
            />
            <VideoGrid videos={results} />
@@ -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) {
                  </div>
                )}
                
+                {/* Verifying Badge - Top Right */}
+                {video.isVerifying && (
+                  <div className="absolute top-2 right-2 z-10">
+                    <Badge variant="secondary" className="text-xs backdrop-blur-md bg-yellow-500/90 text-white animate-pulse">
+                      验证中
+                    </Badge>
+                  </div>
+                )}
+                
                {/* Overlay */}
                <div className="absolute inset-0 bg-gradient-to-t from-black/80 via-black/20 to-transparent opacity-0 group-hover:opacity-100 transition-opacity duration-300">
                  <div className="absolute bottom-0 left-0 right-0 p-3">
@@ -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<void>;
+  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<Video[]>([]);
+  const [availableSources, setAvailableSources] = useState<any[]>([]);
+  const [completedSources, setCompletedSources] = useState(0);
+  const [totalSources, setTotalSources] = useState(0);
+  const [totalVideosFound, setTotalVideosFound] = useState(0);
+  
+  const abortControllerRef = useRef<AbortController | null>(null);
+  const verificationQueueRef = useRef<Video[]>([]);
+  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<string, { count: number; name: string }>();
+
+      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,
+  };
+}
@@ -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'
 ];