(self, query: str, screenshot_dir_path: str)
| 97 | self.api_wrapper = RapidAPI(rapidapi_name) |
| 98 | |
| 99 | async def __call__(self, query: str, screenshot_dir_path: str) -> List[Dict[str, Any]]: |
| 100 | try: |
| 101 | output = self.api_wrapper.query(query, max_results=self.max_results+20) # account for error website |
| 102 | except Exception as e: |
| 103 | logger.error(f"DDGSQueryRun call failed: {e}") |
| 104 | output = [] |
| 105 | |
| 106 | evidences = [] |
| 107 | for idx, result in enumerate(output): |
| 108 | evidence = { |
| 109 | "title": result["title"], |
| 110 | "snippet": result.get("description", result.get("body", "")), |
| 111 | 'url': result['href'], |
| 112 | 'screenshot_path': os.path.join(screenshot_dir_path, f"{idx}.jpg") |
| 113 | } |
| 114 | success = await take_screenshot_async(evidence['url'], os.path.join(screenshot_dir_path, f"{idx}.jpg")) |
| 115 | if success: |
| 116 | evidences.append(evidence) |
| 117 | if len(evidences) == self.max_results: |
| 118 | break |
| 119 | |
| 120 | if not evidences: |
| 121 | evidences = None |
| 122 | |
| 123 | return evidences |
| 124 | ## Screenshot of top section. Set the size to be 1024*1024 |
| 125 | async def take_screenshot_async(url: str, screenshot_path: str, timeout: int = BRIEF_TIMEOUT): |
| 126 | async with async_playwright() as p: |
nothing calls this directly
no test coverage detected