7b5af16f6d
- 9 mm* skills (orchestrator, article, social, publish, analytics) - 6 dependency skills (content-writer, geo-optimizer, etc.) - 3 analytics scripts (GSC, Google Ads, Meta Ads) - Config template + setup guide - SOUL-MM.md persona extension - OrbitOS integration reference
320 lines
10 KiB
Python
320 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
GSC Analytics for moreminimore-service.
|
|
Reads clients-config.json, pulls GSC data using service account key.
|
|
|
|
Usage:
|
|
python3 mm_gsc.py --client moreminimore --days 90
|
|
python3 mm_gsc.py --client moreminimore --days 28
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.request
|
|
import urllib.error
|
|
from datetime import date, timedelta
|
|
|
|
|
|
def load_config():
|
|
"""Load clients-config.json"""
|
|
config_path = os.path.expanduser("~/vault/99_System/clients-config.json")
|
|
if not os.path.exists(config_path):
|
|
print(f"ERROR: Config not found at {config_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
with open(config_path) as f:
|
|
return json.load(f)
|
|
|
|
|
|
def find_client(config, client_id):
|
|
"""Find client by ID"""
|
|
for client in config.get("clients", []):
|
|
if client["id"] == client_id:
|
|
return client
|
|
return None
|
|
|
|
|
|
def get_gsc_property(client):
|
|
"""Get GSC property from client config"""
|
|
analytics = client.get("analytics", {})
|
|
gsc_property = analytics.get("gsc_property", "")
|
|
if not gsc_property:
|
|
print(f"ERROR: No gsc_property set for client '{client['id']}'", file=sys.stderr)
|
|
sys.exit(1)
|
|
return gsc_property
|
|
|
|
|
|
def get_access_token(key_path):
|
|
"""Get access token from service account key using openssl for JWT signing"""
|
|
import subprocess
|
|
import base64
|
|
import time
|
|
|
|
if not os.path.exists(key_path):
|
|
print(f"ERROR: Service account key not found: {key_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
try:
|
|
with open(key_path) as f:
|
|
key_data = json.load(f)
|
|
|
|
# JWT header
|
|
header = {"alg": "RS256", "typ": "JWT"}
|
|
header_b64 = base64.urlsafe_b64encode(json.dumps(header).encode()).decode().rstrip('=')
|
|
|
|
# JWT payload
|
|
now = int(time.time())
|
|
payload = {
|
|
"iss": key_data["client_email"],
|
|
"scope": "https://www.googleapis.com/auth/webmasters.readonly",
|
|
"aud": "https://oauth2.googleapis.com/token",
|
|
"iat": now,
|
|
"exp": now + 3600,
|
|
}
|
|
payload_b64 = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip('=')
|
|
|
|
# Sign with openssl
|
|
signing_input = f"{header_b64}.{payload_b64}"
|
|
|
|
# Write private key to temp file
|
|
import tempfile
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.pem', delete=False) as kf:
|
|
kf.write(key_data["private_key"])
|
|
key_file = kf.name
|
|
|
|
try:
|
|
# Sign with openssl
|
|
result = subprocess.run(
|
|
["openssl", "dgst", "-sha256", "-sign", key_file],
|
|
input=signing_input.encode(),
|
|
capture_output=True,
|
|
timeout=10,
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
print(f"ERROR: openssl signing failed: {result.stderr.decode()}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
signature_b64 = base64.urlsafe_b64encode(result.stdout).decode().rstrip('=')
|
|
signed_jwt = f"{signing_input}.{signature_b64}"
|
|
finally:
|
|
os.unlink(key_file)
|
|
|
|
# Exchange JWT for access token
|
|
token_url = "https://oauth2.googleapis.com/token"
|
|
data = urllib.parse.urlencode({
|
|
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
|
|
"assertion": signed_jwt,
|
|
}).encode()
|
|
|
|
req = urllib.request.Request(token_url, data=data, method="POST")
|
|
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
|
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
token_data = json.loads(resp.read().decode())
|
|
return token_data["access_token"]
|
|
|
|
except Exception as e:
|
|
print(f"ERROR: Failed to get access token: {e}", file=sys.stderr)
|
|
print(f" Key file: {key_path}", file=sys.stderr)
|
|
sys.exit(1)
|
|
def gsc_query(site_property, access_token, request_body):
|
|
"""Execute a GSC API query"""
|
|
# URL encode the site property
|
|
import urllib.parse
|
|
site_encoded = urllib.parse.quote(site_property, safe='')
|
|
|
|
url = f"https://searchconsole.googleapis.com/webmasters/v3/sites/{site_encoded}/searchAnalytics/query"
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
body = json.dumps(request_body).encode()
|
|
req = urllib.request.Request(url, data=body, headers=headers, method="POST")
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
except urllib.error.HTTPError as e:
|
|
error_body = e.read().decode() if e.fp else ""
|
|
print(f"ERROR: GSC API returned {e.code}", file=sys.stderr)
|
|
if error_body:
|
|
try:
|
|
error_json = json.loads(error_body)
|
|
print(json.dumps(error_json, indent=2), file=sys.stderr)
|
|
except:
|
|
print(error_body[:500], file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def get_top_queries(site_property, access_token, days=28):
|
|
"""Get top queries"""
|
|
end_date = date.today() - timedelta(days=3) # GSC data lags ~3 days
|
|
start_date = end_date - timedelta(days=days)
|
|
|
|
result = gsc_query(site_property, access_token, {
|
|
"startDate": start_date.isoformat(),
|
|
"endDate": end_date.isoformat(),
|
|
"dimensions": ["query"],
|
|
"rowLimit": 50,
|
|
"type": "web",
|
|
})
|
|
|
|
return result.get("rows", [])
|
|
|
|
|
|
def get_top_pages(site_property, access_token, days=28):
|
|
"""Get top pages"""
|
|
end_date = date.today() - timedelta(days=3)
|
|
start_date = end_date - timedelta(days=days)
|
|
|
|
result = gsc_query(site_property, access_token, {
|
|
"startDate": start_date.isoformat(),
|
|
"endDate": end_date.isoformat(),
|
|
"dimensions": ["page"],
|
|
"rowLimit": 50,
|
|
"type": "web",
|
|
})
|
|
|
|
return result.get("rows", [])
|
|
|
|
|
|
def get_summary(site_property, access_token, days=28):
|
|
"""Get overall summary"""
|
|
end_date = date.today() - timedelta(days=3)
|
|
start_date = end_date - timedelta(days=days)
|
|
|
|
result = gsc_query(site_property, access_token, {
|
|
"startDate": start_date.isoformat(),
|
|
"endDate": end_date.isoformat(),
|
|
"dimensions": ["date"],
|
|
"type": "web",
|
|
})
|
|
|
|
rows = result.get("rows", [])
|
|
total_clicks = sum(r.get("clicks", 0) for r in rows)
|
|
total_impressions = sum(r.get("impressions", 0) for r in rows)
|
|
avg_ctr = (total_clicks / total_impressions * 100) if total_impressions > 0 else 0
|
|
avg_position = sum(r.get("position", 0) * r.get("impressions", 0) for r in rows) / total_impressions if total_impressions > 0 else 0
|
|
|
|
return {
|
|
"clicks": total_clicks,
|
|
"impressions": total_impressions,
|
|
"ctr": avg_ctr,
|
|
"position": avg_position,
|
|
"start_date": start_date.isoformat(),
|
|
"end_date": end_date.isoformat(),
|
|
}
|
|
|
|
|
|
def display_summary(summary):
|
|
"""Display summary"""
|
|
print(f"📊 Summary ({summary['start_date']} to {summary['end_date']})")
|
|
print("-" * 60)
|
|
print(f" Clicks: {summary['clicks']:,}")
|
|
print(f" Impressions: {summary['impressions']:,}")
|
|
print(f" CTR: {summary['ctr']:.2f}%")
|
|
print(f" Avg Position: {summary['position']:.1f}")
|
|
|
|
|
|
def display_queries(rows):
|
|
"""Display top queries"""
|
|
if not rows:
|
|
print("\nไม่พบ queries")
|
|
return
|
|
|
|
print(f"\n🔑 Top Queries ({len(rows)})")
|
|
print("-" * 90)
|
|
print(f"{'Query':<45} {'Clicks':>10} {'Impressions':>12} {'CTR':>8} {'Position':>10}")
|
|
print("-" * 90)
|
|
|
|
for row in rows[:30]:
|
|
keys = row.get("keys", [])
|
|
query = keys[0] if keys else ""
|
|
ctr = row.get("ctr", 0) * 100
|
|
|
|
print(f"{query[:43]:<45} {row.get('clicks', 0):>10,} "
|
|
f"{row.get('impressions', 0):>12,} "
|
|
f"{ctr:>7.2f}% "
|
|
f"{row.get('position', 0):>10.1f}")
|
|
|
|
|
|
def display_pages(rows):
|
|
"""Display top pages"""
|
|
if not rows:
|
|
print("\nไม่พบ pages")
|
|
return
|
|
|
|
print(f"\n📄 Top Pages ({len(rows)})")
|
|
print("-" * 90)
|
|
print(f"{'Page':<55} {'Clicks':>10} {'Impressions':>12} {'CTR':>8}")
|
|
print("-" * 90)
|
|
|
|
for row in rows[:20]:
|
|
keys = row.get("keys", [])
|
|
page = keys[0] if keys else ""
|
|
ctr = row.get("ctr", 0) * 100
|
|
|
|
print(f"{page[:53]:<55} {row.get('clicks', 0):>10,} "
|
|
f"{row.get('impressions', 0):>12,} "
|
|
f"{ctr:>7.2f}%")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="GSC Analytics for moreminimore-service")
|
|
parser.add_argument("--client", required=True, help="Client ID from clients-config.json")
|
|
parser.add_argument("--days", type=int, default=90, help="Number of days to analyze (default: 90)")
|
|
parser.add_argument("--json", action="store_true", help="Output raw JSON")
|
|
args = parser.parse_args()
|
|
|
|
# Load config
|
|
config = load_config()
|
|
|
|
# Find client
|
|
client = find_client(config, args.client)
|
|
if not client:
|
|
print(f"ERROR: Client '{args.client}' not found in config", file=sys.stderr)
|
|
print(f"Available clients: {[c['id'] for c in config.get('clients', [])]}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Get GSC property
|
|
gsc_property = get_gsc_property(client)
|
|
|
|
# Get service account key
|
|
key_path = config.get("global", {}).get("google", {}).get("service_account_key", "")
|
|
if not key_path:
|
|
print("ERROR: No service_account_key in global.google config", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print(f"📊 GSC Analytics for: {client['display_name']}")
|
|
print(f" Property: {gsc_property}")
|
|
print(f" Period: {args.days} days")
|
|
print()
|
|
|
|
# Get access token
|
|
access_token = get_access_token(os.path.expanduser(key_path))
|
|
|
|
# Get data
|
|
summary = get_summary(gsc_property, access_token, args.days)
|
|
queries = get_top_queries(gsc_property, access_token, args.days)
|
|
pages = get_top_pages(gsc_property, access_token, args.days)
|
|
|
|
if args.json:
|
|
result = {
|
|
"summary": summary,
|
|
"queries": queries,
|
|
"pages": pages,
|
|
}
|
|
print(json.dumps(result, indent=2))
|
|
else:
|
|
display_summary(summary)
|
|
display_queries(queries)
|
|
display_pages(pages)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|