123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""Google Search Console - read API.
|
|
|
|
Search Analytics, Sites and Sitemaps live under the Webmasters v3 API; URL
|
|
Inspection lives under searchconsole.googleapis.com/v1. Request/response bodies
|
|
are forwarded as-is.
|
|
|
|
The site URL (e.g. ``https://example.com/`` or ``sc-domain:example.com``) is
|
|
passed as a query parameter and URL-encoded into the upstream path - this keeps
|
|
our routes clean and avoids ambiguity with the slashes/colons it contains.
|
|
|
|
Credentials: X-GSC-Access-Token (preferred) or X-GSC-Credentials.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
|
|
from fastapi import APIRouter, Body, Depends, Query
|
|
|
|
from .. import config
|
|
from ..clients.google import GoogleApiClient
|
|
from ..credentials import GoogleCredentials, get_gsc_credentials
|
|
|
|
router = APIRouter(prefix="/gsc", tags=["google-search-console"])
|
|
|
|
_QUERY_EXAMPLE = {
|
|
"startDate": "2026-05-01",
|
|
"endDate": "2026-05-31",
|
|
"dimensions": ["query", "page"],
|
|
"rowLimit": 100,
|
|
}
|
|
|
|
_SITE_URL_DESC = (
|
|
"Property in Search Console: a URL-prefix property (e.g. "
|
|
"https://example.com/) or a domain property (e.g. sc-domain:example.com)."
|
|
)
|
|
|
|
|
|
def _client(creds: GoogleCredentials) -> GoogleApiClient:
|
|
return GoogleApiClient(creds, service_name="Search Console")
|
|
|
|
|
|
def _site(site_url: str) -> str:
|
|
# The siteUrl is a single path segment and must be fully URL-encoded.
|
|
return quote(site_url.strip(), safe="")
|
|
|
|
|
|
@router.post(
|
|
"/searchAnalytics/query",
|
|
summary="Query Search Console search traffic (clicks, impressions, CTR, position)",
|
|
)
|
|
async def search_analytics_query(
|
|
site_url: str = Query(..., alias="siteUrl", description=_SITE_URL_DESC),
|
|
body: dict[str, Any] = Body(..., examples=[_QUERY_EXAMPLE]),
|
|
creds: GoogleCredentials = Depends(get_gsc_credentials),
|
|
) -> Any:
|
|
url = f"{config.GSC_DATA_BASE_URL}/sites/{_site(site_url)}/searchAnalytics/query"
|
|
return await _client(creds).post(url, body)
|
|
|
|
|
|
@router.get("/sites", summary="List sites in the account")
|
|
async def list_sites(
|
|
creds: GoogleCredentials = Depends(get_gsc_credentials),
|
|
) -> Any:
|
|
return await _client(creds).get(f"{config.GSC_DATA_BASE_URL}/sites")
|
|
|
|
|
|
@router.get("/site", summary="Get a single site's info and permission level")
|
|
async def get_site(
|
|
site_url: str = Query(..., alias="siteUrl", description=_SITE_URL_DESC),
|
|
creds: GoogleCredentials = Depends(get_gsc_credentials),
|
|
) -> Any:
|
|
return await _client(creds).get(
|
|
f"{config.GSC_DATA_BASE_URL}/sites/{_site(site_url)}"
|
|
)
|
|
|
|
|
|
@router.get("/sitemaps", summary="List sitemaps submitted for a site")
|
|
async def list_sitemaps(
|
|
site_url: str = Query(..., alias="siteUrl", description=_SITE_URL_DESC),
|
|
creds: GoogleCredentials = Depends(get_gsc_credentials),
|
|
) -> Any:
|
|
return await _client(creds).get(
|
|
f"{config.GSC_DATA_BASE_URL}/sites/{_site(site_url)}/sitemaps"
|
|
)
|
|
|
|
|
|
@router.get("/sitemap", summary="Get information about a specific sitemap")
|
|
async def get_sitemap(
|
|
site_url: str = Query(..., alias="siteUrl", description=_SITE_URL_DESC),
|
|
feedpath: str = Query(
|
|
..., description="Full URL of the sitemap, e.g. https://example.com/sitemap.xml"
|
|
),
|
|
creds: GoogleCredentials = Depends(get_gsc_credentials),
|
|
) -> Any:
|
|
url = (
|
|
f"{config.GSC_DATA_BASE_URL}/sites/{_site(site_url)}"
|
|
f"/sitemaps/{quote(feedpath.strip(), safe='')}"
|
|
)
|
|
return await _client(creds).get(url)
|
|
|
|
|
|
@router.post(
|
|
"/urlInspection",
|
|
summary="Inspect the Google index status of a URL",
|
|
)
|
|
async def inspect_url(
|
|
body: dict[str, Any] = Body(
|
|
...,
|
|
examples=[
|
|
{
|
|
"inspectionUrl": "https://example.com/some-page",
|
|
"siteUrl": "https://example.com/",
|
|
"languageCode": "cs",
|
|
}
|
|
],
|
|
description="Requires inspectionUrl and siteUrl; languageCode is optional.",
|
|
),
|
|
creds: GoogleCredentials = Depends(get_gsc_credentials),
|
|
) -> Any:
|
|
url = f"{config.GSC_INSPECT_BASE_URL}/urlInspection/index:inspect"
|
|
return await _client(creds).post(url, body)
|