| """ScrapingBee Google SERP client — rank checking, local SEO, PAA, competitors.""" |
| import os |
| import requests |
| from typing import List, Dict |
|
|
| BASE_URL = "https://app.scrapingbee.com/api/v1/google" |
|
|
|
|
| def _key() -> str: |
| return os.environ.get("SCRAPINGBEE_KEY", "") |
|
|
|
|
| def search(query: str, country_code: str = "sa", language: str = "ar", page: int = 1) -> Dict: |
| """Run a Google search via ScrapingBee. Returns full SERP data.""" |
| key = _key() |
| if not key: |
| return {"error": "no_scrapingbee_key"} |
| try: |
| r = requests.get(BASE_URL, params={ |
| "api_key": key, |
| "search": query, |
| "country_code": country_code, |
| "language": language, |
| "page": page, |
| }, timeout=20) |
| r.raise_for_status() |
| return r.json() |
| except Exception as e: |
| return {"error": str(e)} |
|
|
|
|
| def rank_check(site_url: str, keywords: List[str], country_code: str = "sa", language: str = "ar") -> List[Dict]: |
| """ |
| For each keyword check Google rank of site_url. |
| Returns list of {kw, position, url, title, description, found, top_competitor, paa, local_results}. |
| Caps at 8 keywords to save credits. |
| """ |
| domain = site_url.replace("https://", "").replace("http://", "").rstrip("/").split("/")[0] |
| results = [] |
| for kw in keywords[:8]: |
| data = search(kw, country_code=country_code, language=language) |
| organic = data.get("organic_results", []) |
| found = next((i for i in organic if domain in i.get("url", "")), None) |
| top = organic[0] if organic else None |
| results.append({ |
| "kw": kw, |
| "position": found["position"] if found else None, |
| "url": found["url"] if found else None, |
| "title": found["title"] if found else None, |
| "description": found["description"] if found else None, |
| "found": bool(found), |
| "top_competitor": { |
| "title": top["title"], "url": top["url"], |
| "domain": top.get("domain", ""), "description": top.get("description", "") |
| } if top and not found else None, |
| "paa": data.get("questions", [])[:3], |
| "local_results": data.get("local_results", [])[:3], |
| "ai_overview": data.get("ai_overviews", [])[:1], |
| "related_searches": [r.get("query", r) if isinstance(r, dict) else r |
| for r in data.get("related_searches", [])[:5]], |
| }) |
| return results |
|
|
|
|
| def full_serp_report(site_url: str, keywords: List[str], country_code: str = "sa", language: str = "ar") -> Dict: |
| """ |
| Full SERP report: rank check + competitor landscape for top keyword. |
| """ |
| ranks = rank_check(site_url, keywords, country_code=country_code, language=language) |
|
|
| |
| competitor_landscape = [] |
| if keywords: |
| data = search(keywords[0], country_code=country_code, language=language) |
| domain = site_url.replace("https://", "").replace("http://", "").rstrip("/").split("/")[0] |
| for item in data.get("organic_results", []): |
| competitor_landscape.append({ |
| "position": item["position"], |
| "title": item["title"], |
| "url": item["url"], |
| "domain": item.get("domain", ""), |
| "description": item.get("description", ""), |
| "is_self": domain in item.get("url", ""), |
| "sitelinks": len(item.get("sitelinks", [])), |
| }) |
|
|
| ranked = [r for r in ranks if r["found"]] |
| not_ranked = [r for r in ranks if not r["found"]] |
| avg_pos = round(sum(r["position"] for r in ranked) / len(ranked), 1) if ranked else None |
|
|
| return { |
| "ok": True, |
| "site_url": site_url, |
| "summary": { |
| "keywords_checked": len(ranks), |
| "keywords_ranked": len(ranked), |
| "keywords_not_ranked": len(not_ranked), |
| "avg_position": avg_pos, |
| "top_position": min((r["position"] for r in ranked), default=None), |
| }, |
| "ranks": ranks, |
| "competitor_landscape": competitor_landscape, |
| } |
|
|
