deep_research/scripts/lib/serper_client.py

"""Serper.dev 客户端（Google Search API 代理）。

为什么用 Serper：
- 2500 次免费额度，远超 SerpAPI 的 100/月
- 支持 Google Search、Scholar、News、Images、Maps
- Google Patents 无专用 endpoint，但可用 `site:patents.google.com` 技巧
- 价格比 SerpAPI 便宜 3-5×

用途：
- 专利检索：通用 search + `site:patents.google.com`
- 学术论文：/scholar endpoint
- 新闻：/news endpoint（时效性敏感的行业动态）

httpx 客户端使用 trust_env=False 绕过系统 socks 代理（macOS Clash 会导致 TLS EOF）。
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any, Literal

import httpx


SERPER_BASE = "https://google.serper.dev"


@dataclass
class SerperHit:
    title: str
    url: str
    snippet: str
    source: str = ""       # 论文出处 / 新闻媒体
    date: str = ""         # 发表日期（如 scholar / news 返回的话）
    cited_by: int = 0      # 学术论文的引用数（仅 scholar）


class SerperError(RuntimeError):
    pass


class SerperClient:
    def __init__(self, api_key: str | None = None, timeout: float = 30.0) -> None:
        self.api_key = api_key or os.environ.get("SERPAPI_KEY") or os.environ.get("SERPER_API_KEY")
        if not self.api_key:
            raise SerperError("SERPAPI_KEY / SERPER_API_KEY not set")
        self._client = httpx.Client(trust_env=False, timeout=timeout)

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> "SerperClient":
        return self

    def __exit__(self, *_args: Any) -> None:
        self.close()

    def _post(self, path: str, body: dict) -> dict:
        try:
            r = self._client.post(
                f"{SERPER_BASE}{path}",
                json=body,
                headers={
                    "X-API-KEY": self.api_key,
                    "Content-Type": "application/json",
                },
            )
        except httpx.RequestError as e:
            raise SerperError(f"network error: {e}")
        if r.status_code != 200:
            raise SerperError(f"HTTP {r.status_code}: {r.text[:300]}")
        try:
            return r.json()
        except Exception as e:
            raise SerperError(f"invalid JSON: {e}")

    def search(
        self,
        query: str,
        *,
        num_results: int = 10,
        gl: str = "us",
        hl: str = "en",
    ) -> list[SerperHit]:
        """通用 Google 搜索。支持 site: / filetype: / 引号短语等 Google 高级语法。"""
        data = self._post("/search", {
            "q": query,
            "num": num_results,
            "gl": gl,
            "hl": hl,
        })
        hits: list[SerperHit] = []
        for item in (data.get("organic") or [])[:num_results]:
            hits.append(SerperHit(
                title=(item.get("title") or "")[:200],
                url=item.get("link") or "",
                snippet=(item.get("snippet") or "")[:600],
                date=item.get("date") or "",
            ))
        return hits

    def scholar(
        self,
        query: str,
        *,
        num_results: int = 10,
        year_low: int | None = None,
        year_high: int | None = None,
    ) -> list[SerperHit]:
        """Google Scholar 搜索——学术论文首选。

        返回带引用数、发表年份等元数据，权威信源识别更准确。
        """
        body: dict[str, Any] = {"q": query, "num": num_results}
        if year_low is not None:
            body["tbs"] = f"cdr:1,cd_min:{year_low}" + (f",cd_max:{year_high}" if year_high else "")
        data = self._post("/scholar", body)
        hits: list[SerperHit] = []
        for item in (data.get("organic") or [])[:num_results]:
            hits.append(SerperHit(
                title=(item.get("title") or "")[:200],
                url=item.get("link") or "",
                snippet=(item.get("snippet") or "")[:600],
                source=(item.get("publicationInfo") or "")[:200],
                year=item.get("year") or "",
                cited_by=item.get("citedBy") or 0,
            ) if False else SerperHit(
                title=(item.get("title") or "")[:200],
                url=item.get("link") or "",
                snippet=(item.get("snippet") or "")[:600],
                source=(item.get("publicationInfo") or "")[:200],
                date=str(item.get("year") or ""),
                cited_by=item.get("citedBy") or 0,
            ))
        return hits

    def patents(
        self,
        query: str,
        *,
        num_results: int = 10,
    ) -> list[SerperHit]:
        """Google Patents 检索——用 site: 技巧走通用搜索。

        serper.dev 没有专门的 patents endpoint，但 `site:patents.google.com` 效果很好。
        """
        combined = f"site:patents.google.com {query}"
        return self.search(combined, num_results=num_results)

    def news(
        self,
        query: str,
        *,
        num_results: int = 10,
        time_range: Literal["d", "w", "m", "y"] | None = None,
    ) -> list[SerperHit]:
        """Google News 搜索——时效敏感行业动态。

        time_range: d=24h, w=7d, m=30d, y=1y
        """
        body: dict[str, Any] = {"q": query, "num": num_results}
        if time_range:
            body["tbs"] = f"qdr:{time_range}"
        data = self._post("/news", body)
        hits: list[SerperHit] = []
        for item in (data.get("news") or [])[:num_results]:
            hits.append(SerperHit(
                title=(item.get("title") or "")[:200],
                url=item.get("link") or "",
                snippet=(item.get("snippet") or "")[:600],
                source=(item.get("source") or "")[:200],
                date=item.get("date") or "",
            ))
        return hits


if __name__ == "__main__":
    from scripts.lib.zenmux_client import load_secrets
    load_secrets()
    with SerperClient() as c:
        print("=== Patents: dual-target siRNA ===")
        for h in c.patents("dual-target siRNA GalNAc conjugate", num_results=3):
            print(f"  {h.title[:70]}")
            print(f"    {h.url}")
        print("\n=== Scholar: dual-target RNAi ===")
        for h in c.scholar("dual-target RNAi drug 2024", num_results=3):
            print(f"  {h.title[:70]}  [引用 {h.cited_by}] ({h.date})")
            print(f"    {h.url}")
            print(f"    源: {h.source[:80]}")