Search via Google CSE API. Args: date_restrict: Google CSE dateRestrict value, e.g. "y2" (past 2 years), "m6" (past 6 months), "w4" (past 4 weeks), "d30" (past 30 days). When set, filters results by page indexing date -- no hardcoded
(self, query: str, num: int = 10,
date_restrict: str = "")
| 722 | self.available = bool(api_key and engine_id) |
| 723 | |
| 724 | def search(self, query: str, num: int = 10, |
| 725 | date_restrict: str = "") -> list: |
| 726 | """Search via Google CSE API. |
| 727 | |
| 728 | Args: |
| 729 | date_restrict: Google CSE dateRestrict value, e.g. "y2" (past 2 years), |
| 730 | "m6" (past 6 months), "w4" (past 4 weeks), "d30" (past 30 days). |
| 731 | When set, filters results by page indexing date -- no hardcoded |
| 732 | year strings needed in the query. |
| 733 | """ |
| 734 | if not self.available: |
| 735 | return [] |
| 736 | try: |
| 737 | params = { |
| 738 | "key": self.api_key, "cx": self.engine_id, |
| 739 | "q": query, "num": min(num, 10), "sort": "date", |
| 740 | } |
| 741 | if date_restrict: |
| 742 | params["dateRestrict"] = date_restrict |
| 743 | resp = requests.get(self.BASE_URL, params=params, timeout=15) |
| 744 | resp.raise_for_status() |
| 745 | return resp.json().get("items", []) |
| 746 | except Exception as e: |
| 747 | print(f" ERROR Google Search: {e}") |
| 748 | return [] |
| 749 | |
| 750 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 751 | query = source.get("google_query") |
no test coverage detected