#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 反向搜索引擎模块 (Reverse Search Engine) 支持多平台反向搜索：Google Images, Yandex, TinEye, Bing Visual Search """ import asyncio import aiohttp import hashlib import json import base64 from pathlib import Path from typing import Dict, List, Optional, Any from datetime import datetime from urllib.parse import urljoin, urlencode from video_keyframe_extractor import VideoKeyframeExtractor class ReverseSearchEngine: """反向搜索引擎""" def __init__(self, cache_dir: str = "./cache"): """ 初始化搜索引擎 Args: cache_dir: 缓存目录 """ self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(parents=True, exist_ok=True) self.video_extractor = VideoKeyframeExtractor() # 搜索引擎配置 self.search_engines = { 'google_images': self._search_google_images, 'yandex': self._search_yandex, 'tineye': self._search_tineye, 'bing': self._search_bing } async def search_by_image(self, image_url: str) -> List[Dict[str, Any]]: """ 通过图片进行反向搜索 Args: image_url: 图片URLURL或本地路径 Returns: List[Dict]: 搜索结果列表 """ print(f"正在反向搜索图片: {image_url}") # 检查缓存 cache_key = self._get_cache_key(image_url, "image") cached_result = self._load_from_cache(cache_key) if cached_result: print(f"从缓存加载结果: {image_url}") return cached_result all_results = [] # 并行搜索所有搜索引擎 tasks = [] for engine_name, search_func in self.search_engines.items(): tasks.append(search_func(image_url)) results = await asyncio.gather(*tasks, return_exceptions=True) for engine_name, result in zip(self.search_engines.keys(), results): if isinstance(result, Exception): print(f"{engine_name} 搜索失败: {result}") continue if result: all_results.extend(result) # 保存到缓存 self._save_to_cache(cache_key, all_results) return all_results async def search_by_video(self, video_url: str) -> List[Dict[str, Any]]: """ 通过视频进行反向搜索（提取关键帧后搜索） Args: video_url: 视频URL或本地路径 Returns: List[Dict]: 搜索结果列表 """ print(f"正在提取视频关键帧: {video_url}") # 提取关键帧 keyframes = await self.video_extractor.extract_keyframes(video_url) if not keyframes: print(f"无法提取视频关键帧: {video_url}") return [] print(f"提取到 {len(keyframes)} 个关键帧") # 对每个关键帧进行反向搜索 all_results = [] for keyframe_path in keyframes[:3]: # 只搜索前3个关键帧 keyframe_url = f"file://{keyframe_path.absolute()}" results = await self.search_by_image(keyframe_url) all_results.extend(results) return all_results async def search_by_url(self, url: str) -> List[Dict[str, Any]]: """ 通过URL进行反向搜索 Args: url: 帖子或内容URL Returns: List[Dict]: 搜索结果列表 """ # URL反向搜索通常用于查找相同的URL在不同平台的发布 # 这里简化实现，实际可能需要使用专门的搜索引擎 print(f"正在反向搜索URL: {url}") results = [] # 检查缓存 cache_key = self._get_cache_key(url, "url") cached_result = self._load_from_cache(cache_key) if cached_result: return cached_result # 使用Google搜索相同URL results = await self._search_google_url(url) # 保存到缓存 self._save_to_cache(cache_key, results) return results async def _search_google_images(self, image_url: str) -> List[Dict[str, Any]]: """ Google Images反向搜索 注意：实际实现需要使用Selenium或Playwright，因为Google没有官方API """ print("执行Google Images反向搜索...") # 这里返回模拟结果，实际实现需要： # 1. 上传图片到Google Images # 2. 解析搜索结果页面 # 3. 提取相似图片信息 # 模拟结果 return [ { "engine": "Google Images", "url": "https://example.com/original.jpg", "title": "可能的原始图片", "source_url": "https://example.com/post/123", "platform": "Unknown", "account_name": "user1", "timestamp": "2026-03-30T10:00:00Z", "similarity": 0.95 } ] async def _search_yandex(self, image_url: str) -> List[Dict[str, Any]]: """ Yandex反向搜索（对俄罗斯地区的内容效果好） """ print("执行Yandex反向搜索...") # 模拟结果 return [ { "engine": "Yandex", "url": "https://yandex.com/images/original.jpg", "title": "相似图片", "source_url": "https://vk.com/post/456", "platform": "VK", "account_name": "user2", "timestamp": "2026-03-31T15:30:00Z", "similarity": 0.90 } ] async def _search_tineye(self, image_url: str) -> List[Dict[str, Any]]: """ TinEye反向搜索（专业反向搜索引擎） """ print("执行TinEye反向搜索...") # TinEye有API，但需要API key # 这里返回模拟结果 return [ { "engine": "TinEye", "url": "https://example.com/tineye_match.jpg", "title": "TinEye匹配结果", "source_url": "https://twitter.com/status/789", "platform": "Twitter", "account_name": "@news_account", "timestamp": "2026-04-01T08:15:00Z", "similarity": 0.88 } ] async def _search_bing(self, image_url: str) -> List[Dict[str, Any]]: """ Bing Visual Search """ print("执行Bing Visual Search...") # 模拟结果 return [ { "engine": "Bing", "url": "https://example.com/bing_match.jpg", "title": "Bing匹配结果", "source_url": "https://facebook.com/post/321", "platform": "Facebook", "account_name": "page1", "timestamp": "2026-04-01T12:00:00Z", "similarity": 0.85 } ] async def _search_google_url(self, url: str) -> List[Dict[str, Any]]: """通过Google搜索相同URL""" print("执行Google URL搜索...") # 模拟结果 return [ { "engine": "Google Search", "url": url, "title": "相同链接的帖子", "source_url": "https://example.com/similar-post", "platform": "Unknown", "account_name": "user3", "timestamp": "2026-03-29T09:00:00Z", "similarity": 1.0 } ] def _get_cache_key(self, query: str, search_type: str) -> str: """生成缓存键""" hash_str = hashlib.md5(f"{search_type}:{query}".encode()).hexdigest() return f"{search_type}_{hash_str}" def _load_from_cache(self, cache_key: str) -> Optional[List[Dict[str, Any]]]: """从缓存加载""" cache_file = self.cache_dir / f"{cache_key}.json" if cache_file.exists(): try: with open(cache_file, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"加载缓存失败: {e}") return None def _save_to_cache(self, cache_key: str, results: List[Dict[str, Any]]): """保存到缓存""" cache_file = self.cache_dir / f"{cache_key}.json" try: with open(cache_file, 'w', encoding='utf-8') as f: json.dump(results, f, ensure_ascii=False, indent=2) except Exception as e: print(f"保存缓存失败: {e}")