| 711 | # --------------------------------------------------------------------------- |
| 712 | |
| 713 | class GoogleSearchChecker: |
| 714 | BASE_URL = "https://www.googleapis.com/customsearch/v1" |
| 715 | |
| 716 | def __init__(self, api_key: str, engine_id: str, state: dict, |
| 717 | seed_mode: bool = False): |
| 718 | self.api_key = api_key |
| 719 | self.engine_id = engine_id |
| 720 | self.state = state |
| 721 | self.seed_mode = seed_mode |
| 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") |
| 752 | if not query or not self.available: |
| 753 | if not self.available: |
| 754 | print(f" SKIP (Google CSE not configured)") |
| 755 | return None |
| 756 | title_filter = source.get("title_filter") |
| 757 | prev = self.state.get(source_id, {}) |
| 758 | prev_titles = set(prev.get("seen_titles", [])) |
| 759 | date_restrict = source.get("date_restrict", "") |
| 760 | items = self.search(query, date_restrict=date_restrict) |
| 761 | new_items = [] |
| 762 | filtered_count = 0 |
| 763 | all_titles = list(prev_titles) |
| 764 | for item in items: |
| 765 | title = item.get("title", "") |
| 766 | if title_filter and not re.search(title_filter, title, re.IGNORECASE): |
| 767 | filtered_count += 1 |
| 768 | continue |
| 769 | if title not in prev_titles: |
| 770 | new_items.append({"title": title, "link": item.get("link", ""), |