| import os |
| from typing import Any |
|
|
| import httpx |
| from dotenv import load_dotenv |
|
|
| load_dotenv() |
|
|
| GOOGLE_FC_ENDPOINT = "https://factchecktools.googleapis.com/v1alpha1/claims:search" |
|
|
|
|
| async def search_google_factcheck( |
| query: str, |
| *, |
| language_code: str = "ar", |
| page_size: int = 5, |
| api_key: str | None = None, |
| ) -> list[dict[str, Any]]: |
| resolved_api_key = api_key or os.getenv( |
| "GOOGLE_FACTCHECK_API_KEY", "").strip() |
| if not resolved_api_key: |
| print("[GoogleFC] No API key found in GOOGLE_FACTCHECK_API_KEY.") |
| return [] |
|
|
| params = { |
| "query": query, |
| "key": resolved_api_key, |
| "languageCode": language_code, |
| "pageSize": page_size, |
| } |
|
|
| try: |
| async with httpx.AsyncClient(timeout=10) as client: |
| response = await client.get(GOOGLE_FC_ENDPOINT, params=params) |
| response.raise_for_status() |
| data = response.json() |
| except httpx.HTTPStatusError as exc: |
| print(f"[GoogleFC] HTTP error: {exc.response.status_code}") |
| return [] |
| except Exception as exc: |
| print(f"[GoogleFC] Request error: {exc}") |
| return [] |
|
|
| evidence: list[dict[str, Any]] = [] |
| for claim_item in data.get("claims", []): |
| claim_text = str(claim_item.get("text", "")).strip() |
| for review in claim_item.get("claimReview", []): |
| publisher = str(review.get( |
| "publisher", {}).get("name", "")).strip() |
| rating = str(review.get("textualRating", "")).strip() |
| url = str(review.get("url", "")).strip() |
| title = str(review.get("title", "")).strip() or claim_text |
|
|
| evidence.append( |
| { |
| "snippet": f"[{publisher}] {title} โ ุงูุชูููู
: {rating}", |
| "source_url": url, |
| "source": "google_factcheck", |
| "rating": rating or None, |
| } |
| ) |
|
|
| return evidence |
|
|