""" Reverse Image Search Module 反向图片搜索模块 Supports: Google Images, Yandex, TinEye, Bing Visual Search """ import os import time import base64 import hashlib import mimetypes from typing import List, Dict, Optional from dataclasses import dataclass from pathlib import Path import requests from bs4 import BeautifulSoup from PIL import Image import cv2 import numpy as np from tenacity import retry, stop_after_attempt, wait_exponential @dataclass class SearchResult: """反向搜索结果""" platform: str title: str url: str image_url: str thumbnail_url: Optional[str] = None timestamp: Optional[str] = None similarity: float = 0.0 metadata: Dict = None def __post_init__(self): if self.metadata is None: self.metadata = {} class ReverseImageSearch: """反向图片搜索器""" def __init__(self, timeout: int = 30, user_agent: str = None): """ 初始化反向图片搜索器 Args: timeout: 请求超时时间（秒） user_agent: 自定义User-Agent """ self.timeout = timeout self.user_agent = user_agent or ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ) self.headers = { "User-Agent": self.user_agent, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Accept-Encoding": "gzip, deflate", "Connection": "keep-alive", } self.session = requests.Session() self.session.headers.update(self.headers) def _encode_image_to_base64(self, image_path: str) -> str: """ 将图片编码为base64 Args: image_path: 图片路径 Returns: base64编码的图片数据 """ with open(image_path, "rb") as f: image_data = f.read() base64_data = base64.b64encode(image_data).decode("utf-8") return base64_data def _resize_image_if_needed(self, image_path: str, max_size: int = 1000000) -> str: """ 如果图片太大，则调整大小 Args: image_path: 原始图片路径 max_size: 最大字节数（默认1MB） Returns: 处理后的图片路径（可能是原始路径或临时文件） """ file_size = os.path.getsize(image_path) if file_size <= max_size: return image_path # 需要缩小图片 img = Image.open(image_path) # 计算缩放比例 scale_factor = (max_size / file_size) ** 0.5 new_width = int(img.width * scale_factor) new_height = int(img.height * scale_factor) # 缩放图片 img_resized = img.resize((new_width, new_height), Image.Resampling.LANCZOS) # 保存到临时文件 temp_path = image_path.replace(".", "_resized.") img_resized.save(temp_path, quality=85) return temp_path @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def search_google(self, image_path: str, max_results: int = 10) -> List[SearchResult]: """ 使用Google Images进行反向图片搜索 Args: image_path: 图片路径 max_results: 最大结果数 Returns: 搜索结果列表 """ results = [] try: # 调整图片大小（Google限制上传大小） processed_image = self._resize_image_if_needed(image_path) # 上传图片到Google Images search_url = "https://lens.google.com/upload" files = {"file": open(processed_image, "rb")} response = self.session.post(search_url, files=files, timeout=self.timeout) if response.status_code != 200: raise Exception(f"Google search failed with status {response.status_code}") # 解析搜索结果 # 注意：Google的HTML结构经常变化，这里需要根据实际情况调整 soup = BeautifulSoup(response.text, "html.parser") # 提取搜索结果（示例，实际需要根据Google的HTML结构调整） for item in soup.find_all("div", class_=["hpc", "MkQBvd"]): try: title_elem = item.find("a") if not title_elem: continue title = title_elem.get_text(strip=True) url = title_elem.get("href", "") image_elem = item.find("img") image_url = image_elem.get("src", "") if image_elem else "" if url and image_url: results.append(SearchResult( platform="Google Images", title=title, url=url, image_url=image_url, similarity=0.0 # Google不提供相似度分数 )) if len(results) >= max_results: break except Exception as e: print(f"Error parsing Google result: {e}") continue except Exception as e: print(f"Google image search error: {e}") finally: # 清理临时文件 if "processed_image" in locals() and processed_image != image_path: if os.path.exists(processed_image): os.remove(processed_image) return results @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def search_yandex(self, image_path: str, max_results: int = 10) -> List[SearchResult]: """ 使用Yandex进行反向图片搜索 Args: image_path: 图片路径 max_results: 最大结果数 Returns: 搜索结果列表 """ results = [] try: # Yandex Images API search_url = "https://yandex.com/images/search" params = { "rpt": "imageview", "format": "jpg", "prg": "1" } files = { "upfile": ( os.path.basename(image_path), open(image_path, "rb"), mimetypes.guess_type(image_path)[0] or "image/jpeg" ) } response = self.session.post( search_url, params=params, files=files, timeout=self.timeout ) if response.status_code != 200: raise Exception(f"Yandex search failed with status {response.status_code}") # 解析搜索结果 soup = BeautifulSoup(response.text, "html.parser") # 提取搜索结果 for item in soup.find_all("div", class_=["serp-item", "CbirItem"]): try: link_elem = item.find("a") if not link_elem: continue url = link_elem.get("href", "") title = link_elem.get("title", link_elem.get_text(strip=True)) img_elem = item.find("img") image_url = img_elem.get("src", "") if img_elem else "" if url and image_url: results.append(SearchResult( platform="Yandex", title=title, url=url, image_url=image_url, similarity=0.0 )) if len(results) >= max_results: break except Exception as e: print(f"Error parsing Yandex result: {e}") continue except Exception as e: print(f"Yandex image search error: {e}") return results @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def search_tineye(self, image_path: str, max_results: int = 10) -> List[SearchResult]: """ 使用TinEye进行反向图片搜索 Args: image_path: 图片路径 max_results: 最大结果数 Returns: 搜索结果列表 """ results = [] try: # TinEye API（需要API key，这里使用公共接口） search_url = "https://tineye.com/result/" files = { "image": ( os.path.basename(image_path), open(image_path, "rb"), mimetypes.guess_type(image_path)[0] or "image/jpeg" ) } response = self.session.post(search_url, files=files, timeout=self.timeout) if response.status_code != 200: raise Exception(f"TinEye search failed with status {response.status_code}") # 解析搜索结果 soup = BeautifulSoup(response.text, "html.parser") # 提取搜索结果 for item in soup.find_all("div", class_=["match", "result"]): try: link_elem = item.find("a") if not link_elem: continue url = link_elem.get("href", "") title = link_elem.get_text(strip=True) img_elem = item.find("img") image_url = img_elem.get("src", "") if img_elem else "" # TinEye通常提供时间戳和相似度 time_elem = item.find("span", class_=["date", "time"]) timestamp = time_elem.get_text(strip=True) if time_elem else None similarity_elem = item.find("span", class_=["score"]) similarity = float(similarity_elem.get_text(strip=True).rstrip("%")) / 100.0 if similarity_elem else 0.0 if url: results.append(SearchResult( platform="TinEye", title=title, url=url, image_url=image_url, timestamp=timestamp, similarity=similarity )) if len(results) >= max_results: break except Exception as e: print(f"Error parsing TinEye result: {e}") continue except Exception as e: print(f"TinEye image search error: {e}") return results @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)) def search_bing(self, image_path: str, max_results: int = 10) -> List[SearchResult]: """ 使用Bing Visual Search进行反向图片搜索 Args: image_path: 图片路径 max_results: 最大结果数 Returns: 搜索结果列表 """ results = [] try: # Bing Visual Search search_url = "https://www.bing.com/visualsearch" params = {"form": "VSVH1PC"} files = { "imgfile": ( os.path.basename(image_path), open(image_path, "rb"), mimetypes.guess_type(image_path)[0] or "image/jpeg" ) } response = self.session.post( search_url, params=params, files=files, timeout=self.timeout ) if response.status_code != 200: raise Exception(f"Bing search failed with status {response.status_code}") # 解析搜索结果 soup = BeautifulSoup(response.text, "html.parser") # 提取搜索结果 for item in soup.find_all("div", class_=["richMedia", "mimg"]): try: link_elem = item.find("a") if not link_elem: continue url = link_elem.get("href", "") title = link_elem.get("title", link_elem.get_text(strip=True)) img_elem = item.find("img") image_url = img_elem.get("src", "") if img_elem else "" if url and image_url: results.append(SearchResult( platform="Bing Visual Search", title=title, url=url, image_url=image_url, similarity=0.0 )) if len(results) >= max_results: break except Exception as e: print(f"Error parsing Bing result: {e}") continue except Exception as e: print(f"Bing image search error: {e}") return results def search_all( self, image_path: str, engines: List[str] = None, max_results_per_engine: int = 10 ) -> List[SearchResult]: """ 在所有支持的平台上进行反向图片搜索 Args: image_path: 图片路径 engines: 搜索引擎列表，默认为["google", "yandex", "tineye", "bing"] max_results_per_engine: 每个引擎的最大结果数 Returns: 所有搜索结果的合并列表 """ if engines is None: engines = ["google", "yandex", "tineye", "bing"] all_results = [] # 并行搜索各平台 for engine in engines: try: if engine == "google": results = self.search_google(image_path, max_results_per_engine) elif engine == "yandex": results = self.search_yandex(image_path, max_results_per_engine) elif engine == "tineye": results = self.search_tineye(image_path, max_results_per_engine) elif engine == "bing": results = self.search_bing(image_path, max_results_per_engine) else: print(f"Unknown search engine: {engine}") continue all_results.extend(results) time.sleep(1) # 避免请求过于频繁 except Exception as e: print(f"Error searching with {engine}: {e}") continue # 去重（基于URL） seen_urls = set() unique_results = [] for result in all_results: if result.url not in seen_urls: seen_urls.add(result.url) unique_results.append(result) return unique_results def close(self): """关闭会话""" self.session.close() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close()