Initial commit: moreminimore-service content pipeline
- 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
This commit is contained in:
Kunthawat Greethong
319
scripts/mm_gsc.py
Normal file
319
scripts/mm_gsc.py
Normal file
@@ -0,0 +1,319 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user