Coverage for web_search / providers / google.py: 47%

1import os 

2import time 

3from typing import List 

4from googleapiclient.discovery import build 

5from qrclaw.web_search.types import WebSearchProvider, SearchResult, WebSearchResponse 

6from qrclaw.logger import get_logger 

7 

8# 环境变量：GOOGLE_API_KEY 和 GOOGLE_CSE_ID 

9GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY") 

10GOOGLE_CSE_ID = os.getenv("GOOGLE_CSE_ID") 

11 

12logger = get_logger("qrclaw.web_search.providers.google") 

13 

14class GoogleSearchProvider(WebSearchProvider): 

15 @property 

16 def id(self) -> str: 

17 return "google" 

18 

19 @property 

20 def name(self) -> str: 

21 return "Google Search" 

22 

23 def is_available(self) -> bool: 

24 return bool(GOOGLE_API_KEY and GOOGLE_CSE_ID) 

25 

26 def search(self, query: str, max_results: int = 5, **kwargs) -> WebSearchResponse: 

27 if not self.is_available(): 

28 raise RuntimeError("GOOGLE_API_KEY or GOOGLE_CSE_ID not configured") 

29 

30 start_time = time.time() 

31 logger.debug(f"Google searching: {query}") 

32 

33 try: 

34 service = build("customsearch", "v1", developerKey=GOOGLE_API_KEY) 

35 # max results per page is 10 

36 num = max(1, min(10, int(max_results))) 

37 

38 res = service.cse().list( 

39 q=query, 

40 cx=GOOGLE_CSE_ID, 

41 num=num 

42 ).execute() 

43 

44 items = res.get("items", []) 

45 results: List[SearchResult] = [] 

46 

47 for item in items: 

48 results.append(SearchResult( 

49 title=item.get("title", ""), 

50 url=item.get("link", ""), 

51 snippet=item.get("snippet", "") 

52 )) 

53 

54 except Exception as e: 

55 logger.error(f"Google search failed: {e}") 

56 raise RuntimeError(f"Google API error: {e}") 

57 

58 took_ms = (time.time() - start_time) * 1000 

59 return WebSearchResponse( 

60 query=query, 

61 provider=self.id, 

62 count=len(results), 

63 took_ms=took_ms, 

64 results=results 

65 )