auto-claude: 2.1 - Create scripts/pagespeed_client.py with GooglePageSpeedClient
Implements Google PageSpeed Insights API client with: - GooglePageSpeedClient class for making API calls - Exponential backoff retry logic (3 retries, 1-60s backoff) - RateLimiter class with daily quota tracking (25k req/day) - Quota persistence to .pagespeed_quota.json - Support for mobile/desktop strategies - Core Web Vitals extraction (LCP, FCP, CLS, TTFB) - Lighthouse audit scores (performance, accessibility, SEO, best-practices) - Structured dataclasses for results (PageSpeedResult, PageSpeedScore, CoreWebVitals) - Custom exceptions (QuotaExceededError, RateLimitError, PageSpeedAPIError) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
5ef2752e1a
commit
9f58e3f8e1
4
.gitignore
vendored
4
.gitignore
vendored
@ -46,6 +46,10 @@ deploy_config.conf
|
||||
# Auto Claude data directory
|
||||
.auto-claude/
|
||||
|
||||
# PageSpeed quota tracking files
|
||||
.pagespeed_quota.json
|
||||
scripts/.pagespeed_quota.json
|
||||
|
||||
# Analysis reports and temp files
|
||||
*.csv
|
||||
*_analysis*.json
|
||||
|
||||
741
scripts/pagespeed_client.py
Normal file
741
scripts/pagespeed_client.py
Normal file
@ -0,0 +1,741 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Google PageSpeed Insights API Client
|
||||
=====================================
|
||||
|
||||
Client for interacting with Google PageSpeed Insights API with built-in:
|
||||
- Rate limiting (25,000 requests/day free tier)
|
||||
- Exponential backoff retry logic
|
||||
- Comprehensive error handling
|
||||
|
||||
Usage:
|
||||
from pagespeed_client import GooglePageSpeedClient
|
||||
|
||||
client = GooglePageSpeedClient()
|
||||
result = client.analyze_url('https://example.com')
|
||||
|
||||
Author: Claude Code
|
||||
Date: 2026-01-08
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
import time
|
||||
import logging
|
||||
from datetime import datetime, date
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any, List
|
||||
from dataclasses import dataclass, field, asdict
|
||||
from enum import Enum
|
||||
|
||||
import requests
|
||||
|
||||
# Configure logging
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# API Configuration
|
||||
PAGESPEED_API_URL = 'https://www.googleapis.com/pagespeedonline/v5/runPagespeed'
|
||||
PAGESPEED_API_KEY = os.getenv('GOOGLE_PAGESPEED_API_KEY', '')
|
||||
|
||||
# Rate limiting configuration
|
||||
DAILY_QUOTA_LIMIT = 25000 # Free tier limit
|
||||
REQUESTS_PER_MINUTE = 60 # Conservative limit to avoid bursts
|
||||
MIN_REQUEST_INTERVAL = 1.0 # Minimum seconds between requests
|
||||
|
||||
# Retry configuration
|
||||
MAX_RETRIES = 3
|
||||
INITIAL_BACKOFF = 1.0 # Initial backoff in seconds
|
||||
MAX_BACKOFF = 60.0 # Maximum backoff in seconds
|
||||
BACKOFF_MULTIPLIER = 2.0
|
||||
|
||||
# Request configuration
|
||||
REQUEST_TIMEOUT = 60 # PageSpeed analysis can take a while
|
||||
USER_AGENT = 'NordaBiznes-SEO-Auditor/1.0'
|
||||
|
||||
|
||||
class Strategy(Enum):
|
||||
"""PageSpeed analysis strategy (device type)."""
|
||||
MOBILE = 'mobile'
|
||||
DESKTOP = 'desktop'
|
||||
|
||||
|
||||
class Category(Enum):
|
||||
"""PageSpeed Lighthouse audit categories."""
|
||||
PERFORMANCE = 'performance'
|
||||
ACCESSIBILITY = 'accessibility'
|
||||
BEST_PRACTICES = 'best-practices'
|
||||
SEO = 'seo'
|
||||
|
||||
|
||||
@dataclass
|
||||
class PageSpeedScore:
|
||||
"""Container for PageSpeed Lighthouse scores."""
|
||||
performance: Optional[int] = None
|
||||
accessibility: Optional[int] = None
|
||||
best_practices: Optional[int] = None
|
||||
seo: Optional[int] = None
|
||||
|
||||
def to_dict(self) -> Dict[str, Optional[int]]:
|
||||
return asdict(self)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CoreWebVitals:
|
||||
"""Core Web Vitals metrics from PageSpeed."""
|
||||
lcp_ms: Optional[int] = None # Largest Contentful Paint
|
||||
fid_ms: Optional[int] = None # First Input Delay
|
||||
cls: Optional[float] = None # Cumulative Layout Shift
|
||||
fcp_ms: Optional[int] = None # First Contentful Paint
|
||||
ttfb_ms: Optional[int] = None # Time to First Byte
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return asdict(self)
|
||||
|
||||
|
||||
@dataclass
|
||||
class PageSpeedResult:
|
||||
"""Complete PageSpeed analysis result."""
|
||||
url: str
|
||||
final_url: str
|
||||
strategy: str
|
||||
analyzed_at: datetime
|
||||
scores: PageSpeedScore
|
||||
core_web_vitals: CoreWebVitals
|
||||
audits: Dict[str, Any] = field(default_factory=dict)
|
||||
lighthouse_version: Optional[str] = None
|
||||
fetch_time_ms: Optional[int] = None
|
||||
error: Optional[str] = None
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
result = {
|
||||
'url': self.url,
|
||||
'final_url': self.final_url,
|
||||
'strategy': self.strategy,
|
||||
'analyzed_at': self.analyzed_at.isoformat() if self.analyzed_at else None,
|
||||
'scores': self.scores.to_dict(),
|
||||
'core_web_vitals': self.core_web_vitals.to_dict(),
|
||||
'audits': self.audits,
|
||||
'lighthouse_version': self.lighthouse_version,
|
||||
'fetch_time_ms': self.fetch_time_ms,
|
||||
'error': self.error,
|
||||
}
|
||||
return result
|
||||
|
||||
|
||||
class RateLimiter:
|
||||
"""
|
||||
Simple rate limiter with daily quota tracking.
|
||||
|
||||
Persists quota usage to a JSON file to track usage across script runs.
|
||||
"""
|
||||
|
||||
def __init__(self, daily_limit: int = DAILY_QUOTA_LIMIT,
|
||||
min_interval: float = MIN_REQUEST_INTERVAL,
|
||||
quota_file: Optional[str] = None):
|
||||
self.daily_limit = daily_limit
|
||||
self.min_interval = min_interval
|
||||
self.last_request_time: Optional[float] = None
|
||||
|
||||
# Quota persistence file
|
||||
if quota_file:
|
||||
self.quota_file = Path(quota_file)
|
||||
else:
|
||||
# Default to scripts directory
|
||||
self.quota_file = Path(__file__).parent / '.pagespeed_quota.json'
|
||||
|
||||
self._load_quota()
|
||||
|
||||
def _load_quota(self) -> None:
|
||||
"""Load quota usage from persistent storage."""
|
||||
self.today = date.today().isoformat()
|
||||
self.requests_today = 0
|
||||
|
||||
if self.quota_file.exists():
|
||||
try:
|
||||
with open(self.quota_file, 'r') as f:
|
||||
data = json.load(f)
|
||||
if data.get('date') == self.today:
|
||||
self.requests_today = data.get('requests', 0)
|
||||
else:
|
||||
# New day, reset counter
|
||||
self._save_quota()
|
||||
except (json.JSONDecodeError, IOError) as e:
|
||||
logger.warning(f"Failed to load quota file: {e}")
|
||||
self._save_quota()
|
||||
else:
|
||||
self._save_quota()
|
||||
|
||||
def _save_quota(self) -> None:
|
||||
"""Persist quota usage to file."""
|
||||
try:
|
||||
with open(self.quota_file, 'w') as f:
|
||||
json.dump({
|
||||
'date': self.today,
|
||||
'requests': self.requests_today,
|
||||
'limit': self.daily_limit,
|
||||
}, f)
|
||||
except IOError as e:
|
||||
logger.warning(f"Failed to save quota file: {e}")
|
||||
|
||||
def can_make_request(self) -> bool:
|
||||
"""Check if we can make another request."""
|
||||
# Check daily quota
|
||||
if self.requests_today >= self.daily_limit:
|
||||
return False
|
||||
return True
|
||||
|
||||
def wait_if_needed(self) -> None:
|
||||
"""Wait if necessary to respect rate limits."""
|
||||
if self.last_request_time is not None:
|
||||
elapsed = time.time() - self.last_request_time
|
||||
if elapsed < self.min_interval:
|
||||
sleep_time = self.min_interval - elapsed
|
||||
logger.debug(f"Rate limiting: sleeping {sleep_time:.2f}s")
|
||||
time.sleep(sleep_time)
|
||||
|
||||
def record_request(self) -> None:
|
||||
"""Record that a request was made."""
|
||||
self.last_request_time = time.time()
|
||||
self.requests_today += 1
|
||||
|
||||
# Reset date if it's a new day
|
||||
today = date.today().isoformat()
|
||||
if today != self.today:
|
||||
self.today = today
|
||||
self.requests_today = 1
|
||||
|
||||
self._save_quota()
|
||||
logger.debug(f"Quota: {self.requests_today}/{self.daily_limit} requests today")
|
||||
|
||||
def get_remaining_quota(self) -> int:
|
||||
"""Get remaining requests for today."""
|
||||
return max(0, self.daily_limit - self.requests_today)
|
||||
|
||||
def get_usage_stats(self) -> Dict[str, Any]:
|
||||
"""Get current usage statistics."""
|
||||
return {
|
||||
'date': self.today,
|
||||
'requests_today': self.requests_today,
|
||||
'daily_limit': self.daily_limit,
|
||||
'remaining': self.get_remaining_quota(),
|
||||
'usage_percent': round(self.requests_today / self.daily_limit * 100, 1),
|
||||
}
|
||||
|
||||
|
||||
class PageSpeedAPIError(Exception):
|
||||
"""Base exception for PageSpeed API errors."""
|
||||
pass
|
||||
|
||||
|
||||
class QuotaExceededError(PageSpeedAPIError):
|
||||
"""Raised when daily quota is exceeded."""
|
||||
pass
|
||||
|
||||
|
||||
class RateLimitError(PageSpeedAPIError):
|
||||
"""Raised when API returns 429 Too Many Requests."""
|
||||
pass
|
||||
|
||||
|
||||
class GooglePageSpeedClient:
|
||||
"""
|
||||
Client for Google PageSpeed Insights API.
|
||||
|
||||
Features:
|
||||
- Rate limiting with daily quota tracking
|
||||
- Exponential backoff retry for transient errors
|
||||
- Comprehensive error handling
|
||||
- Support for both mobile and desktop analysis
|
||||
|
||||
Usage:
|
||||
client = GooglePageSpeedClient()
|
||||
|
||||
# Analyze a single URL
|
||||
result = client.analyze_url('https://example.com')
|
||||
|
||||
# Analyze with both mobile and desktop
|
||||
results = client.analyze_url_both_strategies('https://example.com')
|
||||
|
||||
# Check quota before batch processing
|
||||
if client.get_remaining_quota() >= 80:
|
||||
# Process all 80 companies
|
||||
pass
|
||||
"""
|
||||
|
||||
def __init__(self, api_key: Optional[str] = None,
|
||||
rate_limiter: Optional[RateLimiter] = None):
|
||||
"""
|
||||
Initialize PageSpeed client.
|
||||
|
||||
Args:
|
||||
api_key: Google PageSpeed API key. If not provided, uses
|
||||
GOOGLE_PAGESPEED_API_KEY environment variable.
|
||||
rate_limiter: Optional custom rate limiter instance.
|
||||
"""
|
||||
self.api_key = api_key or PAGESPEED_API_KEY
|
||||
if not self.api_key:
|
||||
logger.warning(
|
||||
"No API key provided. PageSpeed API will work but with "
|
||||
"stricter rate limits. Set GOOGLE_PAGESPEED_API_KEY env var."
|
||||
)
|
||||
|
||||
self.rate_limiter = rate_limiter or RateLimiter()
|
||||
self.session = requests.Session()
|
||||
self.session.headers.update({'User-Agent': USER_AGENT})
|
||||
|
||||
def analyze_url(self, url: str,
|
||||
strategy: Strategy = Strategy.MOBILE,
|
||||
categories: Optional[List[Category]] = None) -> PageSpeedResult:
|
||||
"""
|
||||
Analyze a URL using PageSpeed Insights API.
|
||||
|
||||
Args:
|
||||
url: The URL to analyze.
|
||||
strategy: Device strategy (mobile or desktop).
|
||||
categories: List of categories to analyze. Defaults to all.
|
||||
|
||||
Returns:
|
||||
PageSpeedResult with scores and audit details.
|
||||
|
||||
Raises:
|
||||
QuotaExceededError: If daily quota is exhausted.
|
||||
PageSpeedAPIError: For other API errors.
|
||||
"""
|
||||
# Check quota before making request
|
||||
if not self.rate_limiter.can_make_request():
|
||||
raise QuotaExceededError(
|
||||
f"Daily quota of {self.rate_limiter.daily_limit} requests exceeded. "
|
||||
f"Try again tomorrow or use a different API key."
|
||||
)
|
||||
|
||||
# Default to all categories
|
||||
if categories is None:
|
||||
categories = list(Category)
|
||||
|
||||
# Build request parameters
|
||||
params = {
|
||||
'url': url,
|
||||
'strategy': strategy.value,
|
||||
'category': [cat.value for cat in categories],
|
||||
}
|
||||
|
||||
if self.api_key:
|
||||
params['key'] = self.api_key
|
||||
|
||||
# Wait for rate limit
|
||||
self.rate_limiter.wait_if_needed()
|
||||
|
||||
# Make request with retry logic
|
||||
response = self._make_request_with_retry(params)
|
||||
|
||||
# Record successful request
|
||||
self.rate_limiter.record_request()
|
||||
|
||||
# Parse response
|
||||
return self._parse_response(response, url, strategy)
|
||||
|
||||
def analyze_url_both_strategies(self, url: str,
|
||||
categories: Optional[List[Category]] = None
|
||||
) -> Dict[str, PageSpeedResult]:
|
||||
"""
|
||||
Analyze URL for both mobile and desktop strategies.
|
||||
|
||||
Args:
|
||||
url: The URL to analyze.
|
||||
categories: List of categories to analyze.
|
||||
|
||||
Returns:
|
||||
Dict with 'mobile' and 'desktop' PageSpeedResult.
|
||||
"""
|
||||
results = {}
|
||||
|
||||
for strategy in [Strategy.MOBILE, Strategy.DESKTOP]:
|
||||
try:
|
||||
results[strategy.value] = self.analyze_url(url, strategy, categories)
|
||||
except PageSpeedAPIError as e:
|
||||
logger.error(f"Failed to analyze {url} ({strategy.value}): {e}")
|
||||
results[strategy.value] = PageSpeedResult(
|
||||
url=url,
|
||||
final_url=url,
|
||||
strategy=strategy.value,
|
||||
analyzed_at=datetime.now(),
|
||||
scores=PageSpeedScore(),
|
||||
core_web_vitals=CoreWebVitals(),
|
||||
error=str(e),
|
||||
)
|
||||
|
||||
return results
|
||||
|
||||
def _make_request_with_retry(self, params: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Make API request with exponential backoff retry.
|
||||
|
||||
Retries on:
|
||||
- 429 Too Many Requests
|
||||
- 5xx Server Errors
|
||||
- Connection errors
|
||||
|
||||
Args:
|
||||
params: Request parameters.
|
||||
|
||||
Returns:
|
||||
Parsed JSON response.
|
||||
|
||||
Raises:
|
||||
PageSpeedAPIError: If all retries fail.
|
||||
"""
|
||||
last_error: Optional[Exception] = None
|
||||
backoff = INITIAL_BACKOFF
|
||||
|
||||
for attempt in range(MAX_RETRIES + 1):
|
||||
try:
|
||||
logger.debug(f"API request attempt {attempt + 1}/{MAX_RETRIES + 1}")
|
||||
|
||||
response = self.session.get(
|
||||
PAGESPEED_API_URL,
|
||||
params=params,
|
||||
timeout=REQUEST_TIMEOUT,
|
||||
)
|
||||
|
||||
# Handle rate limiting (429)
|
||||
if response.status_code == 429:
|
||||
retry_after = response.headers.get('Retry-After', backoff)
|
||||
try:
|
||||
retry_after = float(retry_after)
|
||||
except ValueError:
|
||||
retry_after = backoff
|
||||
|
||||
if attempt < MAX_RETRIES:
|
||||
logger.warning(
|
||||
f"Rate limited (429). Retrying in {retry_after}s "
|
||||
f"(attempt {attempt + 1}/{MAX_RETRIES + 1})"
|
||||
)
|
||||
time.sleep(retry_after)
|
||||
backoff = min(backoff * BACKOFF_MULTIPLIER, MAX_BACKOFF)
|
||||
continue
|
||||
else:
|
||||
raise RateLimitError(
|
||||
f"Rate limited after {MAX_RETRIES + 1} attempts"
|
||||
)
|
||||
|
||||
# Handle server errors (5xx)
|
||||
if response.status_code >= 500:
|
||||
if attempt < MAX_RETRIES:
|
||||
logger.warning(
|
||||
f"Server error ({response.status_code}). "
|
||||
f"Retrying in {backoff}s "
|
||||
f"(attempt {attempt + 1}/{MAX_RETRIES + 1})"
|
||||
)
|
||||
time.sleep(backoff)
|
||||
backoff = min(backoff * BACKOFF_MULTIPLIER, MAX_BACKOFF)
|
||||
continue
|
||||
else:
|
||||
raise PageSpeedAPIError(
|
||||
f"Server error {response.status_code} after "
|
||||
f"{MAX_RETRIES + 1} attempts"
|
||||
)
|
||||
|
||||
# Handle client errors (4xx except 429)
|
||||
if response.status_code >= 400:
|
||||
error_data = response.json().get('error', {})
|
||||
error_message = error_data.get('message', response.text)
|
||||
raise PageSpeedAPIError(
|
||||
f"API error {response.status_code}: {error_message}"
|
||||
)
|
||||
|
||||
# Success
|
||||
return response.json()
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
last_error = PageSpeedAPIError(
|
||||
f"Request timed out after {REQUEST_TIMEOUT}s"
|
||||
)
|
||||
if attempt < MAX_RETRIES:
|
||||
logger.warning(
|
||||
f"Request timeout. Retrying in {backoff}s "
|
||||
f"(attempt {attempt + 1}/{MAX_RETRIES + 1})"
|
||||
)
|
||||
time.sleep(backoff)
|
||||
backoff = min(backoff * BACKOFF_MULTIPLIER, MAX_BACKOFF)
|
||||
continue
|
||||
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
last_error = PageSpeedAPIError(f"Connection error: {e}")
|
||||
if attempt < MAX_RETRIES:
|
||||
logger.warning(
|
||||
f"Connection error. Retrying in {backoff}s "
|
||||
f"(attempt {attempt + 1}/{MAX_RETRIES + 1})"
|
||||
)
|
||||
time.sleep(backoff)
|
||||
backoff = min(backoff * BACKOFF_MULTIPLIER, MAX_BACKOFF)
|
||||
continue
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
last_error = PageSpeedAPIError(f"Request failed: {e}")
|
||||
if attempt < MAX_RETRIES:
|
||||
logger.warning(
|
||||
f"Request error. Retrying in {backoff}s "
|
||||
f"(attempt {attempt + 1}/{MAX_RETRIES + 1})"
|
||||
)
|
||||
time.sleep(backoff)
|
||||
backoff = min(backoff * BACKOFF_MULTIPLIER, MAX_BACKOFF)
|
||||
continue
|
||||
|
||||
# All retries exhausted
|
||||
raise last_error or PageSpeedAPIError("Request failed after all retries")
|
||||
|
||||
def _parse_response(self, data: Dict[str, Any],
|
||||
original_url: str,
|
||||
strategy: Strategy) -> PageSpeedResult:
|
||||
"""
|
||||
Parse PageSpeed API response into structured result.
|
||||
|
||||
Args:
|
||||
data: Raw API response.
|
||||
original_url: The URL that was analyzed.
|
||||
strategy: The analysis strategy used.
|
||||
|
||||
Returns:
|
||||
PageSpeedResult with parsed data.
|
||||
"""
|
||||
lighthouse = data.get('lighthouseResult', {})
|
||||
|
||||
# Extract scores (0-1 float -> 0-100 int)
|
||||
categories = lighthouse.get('categories', {})
|
||||
scores = PageSpeedScore(
|
||||
performance=self._extract_score(categories.get('performance')),
|
||||
accessibility=self._extract_score(categories.get('accessibility')),
|
||||
best_practices=self._extract_score(categories.get('best-practices')),
|
||||
seo=self._extract_score(categories.get('seo')),
|
||||
)
|
||||
|
||||
# Extract Core Web Vitals
|
||||
audits = lighthouse.get('audits', {})
|
||||
core_web_vitals = CoreWebVitals(
|
||||
lcp_ms=self._extract_metric_ms(audits.get('largest-contentful-paint')),
|
||||
fid_ms=self._extract_metric_ms(audits.get('max-potential-fid')),
|
||||
cls=self._extract_cls(audits.get('cumulative-layout-shift')),
|
||||
fcp_ms=self._extract_metric_ms(audits.get('first-contentful-paint')),
|
||||
ttfb_ms=self._extract_metric_ms(audits.get('server-response-time')),
|
||||
)
|
||||
|
||||
# Extract relevant audits for SEO
|
||||
seo_audits = self._extract_seo_audits(audits)
|
||||
|
||||
# Get timing info
|
||||
timing = lighthouse.get('timing', {})
|
||||
fetch_time = timing.get('total')
|
||||
|
||||
return PageSpeedResult(
|
||||
url=original_url,
|
||||
final_url=lighthouse.get('finalUrl', original_url),
|
||||
strategy=strategy.value,
|
||||
analyzed_at=datetime.now(),
|
||||
scores=scores,
|
||||
core_web_vitals=core_web_vitals,
|
||||
audits=seo_audits,
|
||||
lighthouse_version=lighthouse.get('lighthouseVersion'),
|
||||
fetch_time_ms=int(fetch_time) if fetch_time else None,
|
||||
)
|
||||
|
||||
def _extract_score(self, category_data: Optional[Dict]) -> Optional[int]:
|
||||
"""Extract score from category data (0-1 float -> 0-100 int)."""
|
||||
if not category_data:
|
||||
return None
|
||||
score = category_data.get('score')
|
||||
if score is not None:
|
||||
return int(round(score * 100))
|
||||
return None
|
||||
|
||||
def _extract_metric_ms(self, audit_data: Optional[Dict]) -> Optional[int]:
|
||||
"""Extract metric value in milliseconds."""
|
||||
if not audit_data:
|
||||
return None
|
||||
value = audit_data.get('numericValue')
|
||||
if value is not None:
|
||||
return int(round(value))
|
||||
return None
|
||||
|
||||
def _extract_cls(self, audit_data: Optional[Dict]) -> Optional[float]:
|
||||
"""Extract Cumulative Layout Shift value."""
|
||||
if not audit_data:
|
||||
return None
|
||||
value = audit_data.get('numericValue')
|
||||
if value is not None:
|
||||
return round(value, 3)
|
||||
return None
|
||||
|
||||
def _extract_seo_audits(self, audits: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Extract SEO-relevant audits from Lighthouse results.
|
||||
|
||||
Returns a dict with audit results organized by category.
|
||||
"""
|
||||
seo_audits = {
|
||||
'meta': {},
|
||||
'crawlability': {},
|
||||
'content': {},
|
||||
'mobile': {},
|
||||
'performance': {},
|
||||
}
|
||||
|
||||
# Meta tags
|
||||
meta_audits = [
|
||||
'document-title',
|
||||
'meta-description',
|
||||
'viewport',
|
||||
'hreflang',
|
||||
'canonical',
|
||||
'robots-txt',
|
||||
]
|
||||
for audit_id in meta_audits:
|
||||
if audit_id in audits:
|
||||
audit = audits[audit_id]
|
||||
seo_audits['meta'][audit_id] = {
|
||||
'score': audit.get('score'),
|
||||
'title': audit.get('title'),
|
||||
'description': audit.get('description'),
|
||||
}
|
||||
|
||||
# Crawlability
|
||||
crawl_audits = [
|
||||
'is-crawlable',
|
||||
'http-status-code',
|
||||
'link-text',
|
||||
'crawlable-anchors',
|
||||
]
|
||||
for audit_id in crawl_audits:
|
||||
if audit_id in audits:
|
||||
audit = audits[audit_id]
|
||||
seo_audits['crawlability'][audit_id] = {
|
||||
'score': audit.get('score'),
|
||||
'title': audit.get('title'),
|
||||
}
|
||||
|
||||
# Content
|
||||
content_audits = [
|
||||
'image-alt',
|
||||
'structured-data',
|
||||
'font-size',
|
||||
'tap-targets',
|
||||
]
|
||||
for audit_id in content_audits:
|
||||
if audit_id in audits:
|
||||
audit = audits[audit_id]
|
||||
seo_audits['content'][audit_id] = {
|
||||
'score': audit.get('score'),
|
||||
'title': audit.get('title'),
|
||||
}
|
||||
|
||||
# Mobile
|
||||
mobile_audits = [
|
||||
'viewport',
|
||||
'content-width',
|
||||
]
|
||||
for audit_id in mobile_audits:
|
||||
if audit_id in audits:
|
||||
audit = audits[audit_id]
|
||||
seo_audits['mobile'][audit_id] = {
|
||||
'score': audit.get('score'),
|
||||
'title': audit.get('title'),
|
||||
}
|
||||
|
||||
# Performance (affects SEO)
|
||||
perf_audits = [
|
||||
'speed-index',
|
||||
'interactive',
|
||||
'total-blocking-time',
|
||||
]
|
||||
for audit_id in perf_audits:
|
||||
if audit_id in audits:
|
||||
audit = audits[audit_id]
|
||||
seo_audits['performance'][audit_id] = {
|
||||
'score': audit.get('score'),
|
||||
'numericValue': audit.get('numericValue'),
|
||||
'displayValue': audit.get('displayValue'),
|
||||
}
|
||||
|
||||
return seo_audits
|
||||
|
||||
def get_remaining_quota(self) -> int:
|
||||
"""Get remaining API requests for today."""
|
||||
return self.rate_limiter.get_remaining_quota()
|
||||
|
||||
def get_usage_stats(self) -> Dict[str, Any]:
|
||||
"""Get API usage statistics."""
|
||||
return self.rate_limiter.get_usage_stats()
|
||||
|
||||
|
||||
# Convenience function for simple usage
|
||||
def analyze_url(url: str, strategy: str = 'mobile') -> Dict[str, Any]:
|
||||
"""
|
||||
Convenience function to analyze a URL.
|
||||
|
||||
Args:
|
||||
url: The URL to analyze.
|
||||
strategy: 'mobile' or 'desktop'.
|
||||
|
||||
Returns:
|
||||
Dict with analysis results.
|
||||
"""
|
||||
client = GooglePageSpeedClient()
|
||||
strat = Strategy.MOBILE if strategy == 'mobile' else Strategy.DESKTOP
|
||||
result = client.analyze_url(url, strat)
|
||||
return result.to_dict()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Quick test
|
||||
import sys
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: python pagespeed_client.py <url>")
|
||||
print("Example: python pagespeed_client.py https://pixlab.pl")
|
||||
sys.exit(1)
|
||||
|
||||
test_url = sys.argv[1]
|
||||
|
||||
print(f"Analyzing: {test_url}")
|
||||
print("-" * 60)
|
||||
|
||||
client = GooglePageSpeedClient()
|
||||
|
||||
print(f"API Key: {'Set' if client.api_key else 'Not set (using public API)'}")
|
||||
print(f"Remaining quota: {client.get_remaining_quota()}")
|
||||
print("-" * 60)
|
||||
|
||||
try:
|
||||
result = client.analyze_url(test_url)
|
||||
|
||||
print(f"URL: {result.url}")
|
||||
print(f"Final URL: {result.final_url}")
|
||||
print(f"Strategy: {result.strategy}")
|
||||
print(f"Analyzed at: {result.analyzed_at}")
|
||||
print()
|
||||
print("Scores:")
|
||||
print(f" Performance: {result.scores.performance}")
|
||||
print(f" Accessibility: {result.scores.accessibility}")
|
||||
print(f" Best Practices: {result.scores.best_practices}")
|
||||
print(f" SEO: {result.scores.seo}")
|
||||
print()
|
||||
print("Core Web Vitals:")
|
||||
print(f" LCP: {result.core_web_vitals.lcp_ms}ms")
|
||||
print(f" FCP: {result.core_web_vitals.fcp_ms}ms")
|
||||
print(f" CLS: {result.core_web_vitals.cls}")
|
||||
print(f" TTFB: {result.core_web_vitals.ttfb_ms}ms")
|
||||
print()
|
||||
print(f"Lighthouse version: {result.lighthouse_version}")
|
||||
print(f"Fetch time: {result.fetch_time_ms}ms")
|
||||
print()
|
||||
print(f"Remaining quota: {client.get_remaining_quota()}")
|
||||
|
||||
except QuotaExceededError as e:
|
||||
print(f"ERROR: Quota exceeded - {e}")
|
||||
sys.exit(1)
|
||||
except PageSpeedAPIError as e:
|
||||
print(f"ERROR: API error - {e}")
|
||||
sys.exit(1)
|
||||
Loading…
Reference in New Issue
Block a user