354 lines
14 KiB
Python
354 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Perfect Postcode: weekly growth-metrics readout.
|
|
|
|
Assembles a one-screen, dated markdown block (SEO + YouTube + product funnel)
|
|
suitable for pasting into a weekly note. Every section degrades gracefully: if a
|
|
credential is missing the section prints "⚠ set X to enable" instead of crashing.
|
|
|
|
USAGE
|
|
python3 analysis/weekly_readout.py # prints the readout to stdout
|
|
|
|
DATA SOURCES & CREDENTIALS (all read from the environment)
|
|
|
|
SEO: Google Search Console (Search Analytics API)
|
|
GSC_SITE_URL Property as registered in GSC, e.g.
|
|
"sc-domain:perfect-postcode.co.uk" or
|
|
"https://perfect-postcode.co.uk/".
|
|
GOOGLE_APPLICATION_CREDENTIALS (or GSC_CREDENTIALS)
|
|
Path to a service-account JSON key. Create it in
|
|
Google Cloud → IAM → Service Accounts, enable the
|
|
"Google Search Console API", then in Search Console
|
|
→ Settings → Users add the service-account email as a
|
|
(restricted) user on the property.
|
|
|
|
YouTube: view counts (Data API v3, API key) + impressions/CTR/retention
|
|
(Analytics API, founder OAuth)
|
|
YT_API_KEY Public Data API v3 key (Cloud console → Credentials).
|
|
Gives per-video view/like/comment counts.
|
|
YT_CHANNEL_ID Channel id (starts "UC..."). See
|
|
youtube.com/account_advanced.
|
|
YT_OAUTH_TOKEN (optional) Path to an authorized_user OAuth token
|
|
JSON for the channel owner. ONLY this unlocks
|
|
impressions, CTR and average view duration, which are
|
|
private and need the founder's Google login + scope
|
|
yt-analytics.readonly. Without it we show public views
|
|
only.
|
|
|
|
Funnel: Plausible (self-hosted) Stats API v2
|
|
PLAUSIBLE_API_KEY Stats API key (Plausible → Settings → API Keys).
|
|
PLAUSIBLE_SITE_ID (default perfect-postcode.co.uk)
|
|
PLAUSIBLE_HOST (default https://stats.schmelczer.dev)
|
|
|
|
KILL / KEEP rule (printed at the bottom): keep going if BOTH search impressions
|
|
AND map opens grew month-over-month; otherwise reconsider.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import date, timedelta
|
|
|
|
from google.auth.transport.requests import Request
|
|
from google.oauth2 import service_account
|
|
from google.oauth2.credentials import Credentials
|
|
|
|
TODAY = date.today()
|
|
WARN = "⚠"
|
|
|
|
|
|
# Comparison windows (yesterday-anchored). GSC data lags ~2-3 days, so the most
|
|
# recent days of the SEO section may read low, which is expected.
|
|
def window(end_offset: int, length: int) -> tuple[str, str]:
|
|
end = TODAY - timedelta(days=end_offset)
|
|
return (end - timedelta(days=length - 1)).isoformat(), end.isoformat()
|
|
|
|
|
|
WK_CUR, WK_PREV = window(1, 7), window(8, 7) # weekly readout
|
|
MO_CUR, MO_PREV = window(1, 28), window(29, 28) # monthly kill/keep gate
|
|
|
|
|
|
def delta(cur: float, prev: float) -> str:
|
|
if prev == 0:
|
|
return " (new)" if cur else ""
|
|
pct = (cur - prev) / prev * 100
|
|
return f" {'▲' if pct >= 0 else '▼'}{pct:+.0f}%"
|
|
|
|
|
|
def http_json(url, *, method="GET", headers=None, body=None):
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(url, data=data, method=method, headers=headers or {})
|
|
if data is not None:
|
|
req.add_header("Content-Type", "application/json")
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
|
|
def google_token(creds_path, scopes, *, authorized_user):
|
|
"""Mint an OAuth bearer token from a Google credential file."""
|
|
if authorized_user:
|
|
creds = Credentials.from_authorized_user_file(creds_path, scopes)
|
|
else:
|
|
creds = service_account.Credentials.from_service_account_file(
|
|
creds_path, scopes=scopes
|
|
)
|
|
creds.refresh(Request())
|
|
return creds.token
|
|
|
|
|
|
# --- shared GSC + Plausible query helpers (reused by sections and kill/keep) ---
|
|
def gsc_creds():
|
|
return os.environ.get("GSC_SITE_URL"), (
|
|
os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
|
|
or os.environ.get("GSC_CREDENTIALS")
|
|
)
|
|
|
|
|
|
_GSC_TOKEN = {} # cache: creds_path -> token
|
|
|
|
|
|
def gsc_query(start, end, dimensions=None, limit=1):
|
|
"""Search Analytics query. Caller must have verified gsc_creds() first."""
|
|
site, creds = gsc_creds()
|
|
if creds not in _GSC_TOKEN:
|
|
_GSC_TOKEN[creds] = google_token(
|
|
creds,
|
|
["https://www.googleapis.com/auth/webmasters.readonly"],
|
|
authorized_user=False,
|
|
)
|
|
body = {"startDate": start, "endDate": end, "rowLimit": limit}
|
|
if dimensions:
|
|
body["dimensions"] = dimensions
|
|
url = (
|
|
"https://searchconsole.googleapis.com/webmasters/v3/sites/"
|
|
+ urllib.parse.quote(site, safe="")
|
|
+ "/searchAnalytics/query"
|
|
)
|
|
headers = {"Authorization": f"Bearer {_GSC_TOKEN[creds]}"}
|
|
return http_json(url, method="POST", headers=headers, body=body).get("rows", [])
|
|
|
|
|
|
def plausible_query(metrics, date_range, filters=None):
|
|
"""Stats API v2 aggregate query -> list of metric values (aligned to
|
|
`metrics`). Caller must have verified PLAUSIBLE_API_KEY first."""
|
|
key = os.environ["PLAUSIBLE_API_KEY"]
|
|
site = os.environ.get("PLAUSIBLE_SITE_ID", "perfect-postcode.co.uk")
|
|
host = os.environ.get("PLAUSIBLE_HOST", "https://stats.schmelczer.dev").rstrip("/")
|
|
body = {"site_id": site, "metrics": metrics, "date_range": list(date_range)}
|
|
if filters:
|
|
body["filters"] = filters
|
|
res = http_json(
|
|
host + "/api/v2/query",
|
|
method="POST",
|
|
headers={"Authorization": f"Bearer {key}"},
|
|
body=body,
|
|
).get("results", [])
|
|
return res[0]["metrics"] if res else [0] * len(metrics)
|
|
|
|
|
|
# Real Plausible events (frontend/src/lib/analytics.ts + MapPage.tsx):
|
|
# pageview /dashboard -> map opens (no dedicated "Map Open" event exists)
|
|
# "Filter Add" -> a filter was applied (props.feature = feature name)
|
|
# "Upgrade Modal Shown" -> the 3-filter demo cap (DEMO_MAX_FILTERS) was hit
|
|
MAP_FILTER = [["is", "event:page", ["/dashboard"]]]
|
|
ADD_FILTER = [["is", "event:name", ["Filter Add"]]]
|
|
CAP_FILTER = [["is", "event:name", ["Upgrade Modal Shown"]]]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. SEO: Google Search Console
|
|
# ---------------------------------------------------------------------------
|
|
def section_seo() -> None:
|
|
print("## SEO: Google Search Console")
|
|
site, creds = gsc_creds()
|
|
if not site or not creds:
|
|
print(f"{WARN} set GSC_SITE_URL + GOOGLE_APPLICATION_CREDENTIALS to enable\n")
|
|
return
|
|
try:
|
|
cur = gsc_query(*WK_CUR)
|
|
prev = gsc_query(*WK_PREV)
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"{WARN} GSC failed: {exc}\n")
|
|
return
|
|
|
|
c, p = (cur[0] if cur else {}), (prev[0] if prev else {})
|
|
ci, cc = c.get("impressions", 0), c.get("clicks", 0)
|
|
pi, pc = p.get("impressions", 0), p.get("clicks", 0)
|
|
print(f" impressions {ci:>8,.0f}{delta(ci, pi)} (prior {pi:,.0f})")
|
|
print(f" clicks {cc:>8,.0f}{delta(cc, pc)} (prior {pc:,.0f})")
|
|
print(f" CTR {(cc / ci * 100 if ci else 0):>7.2f}%")
|
|
|
|
rows = sorted(
|
|
gsc_query(*WK_CUR, dimensions=["page"], limit=1000),
|
|
key=lambda r: r.get("impressions", 0),
|
|
reverse=True,
|
|
)
|
|
watch = ("twin", "postcode", "dashboard", "property-search", "cheaper")
|
|
print(" top pages (impressions / clicks):")
|
|
for r in rows[:8]:
|
|
page = r["keys"][0]
|
|
star = " ★" if any(w in page.lower() for w in watch) else ""
|
|
path = page.replace("https://perfect-postcode.co.uk", "") or "/"
|
|
print(
|
|
f" {r.get('impressions', 0):>6,.0f} / {r.get('clicks', 0):>4,.0f}"
|
|
f" {path}{star}"
|
|
)
|
|
print()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. YouTube
|
|
# ---------------------------------------------------------------------------
|
|
def section_youtube() -> None:
|
|
print("## YouTube")
|
|
api_key, channel = os.environ.get("YT_API_KEY"), os.environ.get("YT_CHANNEL_ID")
|
|
if not api_key or not channel:
|
|
print(f"{WARN} set YT_API_KEY + YT_CHANNEL_ID to enable\n")
|
|
return
|
|
try:
|
|
search = http_json(
|
|
"https://www.googleapis.com/youtube/v3/search?"
|
|
f"key={api_key}&channelId={channel}&part=id&type=video"
|
|
"&order=date&maxResults=15"
|
|
)
|
|
ids = [
|
|
it["id"]["videoId"]
|
|
for it in search.get("items", [])
|
|
if it.get("id", {}).get("videoId")
|
|
]
|
|
if not ids:
|
|
print(" (no public videos found)\n")
|
|
return
|
|
stats = http_json(
|
|
"https://www.googleapis.com/youtube/v3/videos?"
|
|
f"key={api_key}&part=snippet,statistics&id={','.join(ids)}"
|
|
)
|
|
print(" public views (Data API v3):")
|
|
for it in stats.get("items", []):
|
|
views = int(it.get("statistics", {}).get("viewCount", 0))
|
|
print(f" {views:>8,} {it['snippet']['title'][:48]}")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"{WARN} YouTube Data API failed: {exc}")
|
|
|
|
oauth = os.environ.get("YT_OAUTH_TOKEN") # private metrics need founder OAuth
|
|
if not oauth:
|
|
print(
|
|
f" {WARN} set YT_OAUTH_TOKEN (founder OAuth) to add impressions, CTR "
|
|
"and avg view duration\n"
|
|
)
|
|
return
|
|
try:
|
|
token = google_token(
|
|
oauth,
|
|
["https://www.googleapis.com/auth/yt-analytics.readonly"],
|
|
authorized_user=True,
|
|
)
|
|
rep = http_json(
|
|
"https://youtubeanalytics.googleapis.com/v2/reports?ids=channel==MINE"
|
|
f"&startDate={WK_CUR[0]}&endDate={WK_CUR[1]}"
|
|
"&metrics=impressions,impressionsClickThroughRate,averageViewDuration,views"
|
|
"&dimensions=video&sort=-impressions&maxResults=15",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
print(" impressions / CTR% / avg-dur(s) (Analytics API):")
|
|
for vid, impr, ctr, avgdur, _views in rep.get("rows", []):
|
|
print(f" {impr:>7,} CTR {ctr:>5.1f}% {avgdur:>4.0f}s {vid}")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f" {WARN} YouTube Analytics failed: {exc}")
|
|
print()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Funnel: Plausible (self-hosted)
|
|
# ---------------------------------------------------------------------------
|
|
def section_funnel():
|
|
print("## Funnel: Plausible")
|
|
if not os.environ.get("PLAUSIBLE_API_KEY"):
|
|
site = os.environ.get("PLAUSIBLE_SITE_ID", "perfect-postcode.co.uk")
|
|
print(f"{WARN} set PLAUSIBLE_API_KEY to enable (site={site})\n")
|
|
return None
|
|
|
|
def snapshot(rng):
|
|
return (
|
|
plausible_query(["visitors"], rng)[0],
|
|
plausible_query(["pageviews"], rng, MAP_FILTER)[0],
|
|
plausible_query(["visitors"], rng, ADD_FILTER)[0],
|
|
plausible_query(["visitors"], rng, CAP_FILTER)[0],
|
|
)
|
|
|
|
try:
|
|
cv, cm, cf, cc = snapshot(WK_CUR)
|
|
pv, pm, _pf, _pc = snapshot(WK_PREV)
|
|
except urllib.error.HTTPError as exc:
|
|
print(f"{WARN} Plausible query failed ({exc.code}); check key/site_id\n")
|
|
return None
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"{WARN} Plausible query failed: {exc}\n")
|
|
return None
|
|
|
|
print(f" visitors {cv:>7,}{delta(cv, pv)} (prior {pv:,})")
|
|
print(f" map opens {cm:>7,}{delta(cm, pm)} (prior {pm:,})")
|
|
print(
|
|
f" ≥1 filter applied {cf:>7,} = {(cf / cv * 100 if cv else 0):.0f}% of visitors"
|
|
)
|
|
print(
|
|
f" 3-filter cap hit {cc:>7,} = {(cc / cm * 100 if cm else 0):.0f}% of map opens"
|
|
)
|
|
print()
|
|
return cm # weekly map opens (unused below, but handy for callers)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Kill / keep: month-over-month gate
|
|
# ---------------------------------------------------------------------------
|
|
def kill_keep(_map_opens_weekly) -> None:
|
|
print("## Kill / Keep")
|
|
impr_up = map_up = None
|
|
|
|
site, creds = gsc_creds()
|
|
if site and creds:
|
|
try:
|
|
cur = gsc_query(*MO_CUR)
|
|
prev = gsc_query(*MO_PREV)
|
|
impr_up = (cur[0]["impressions"] if cur else 0) > (
|
|
prev[0]["impressions"] if prev else 0
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
impr_up = None
|
|
|
|
if os.environ.get("PLAUSIBLE_API_KEY"):
|
|
try:
|
|
map_up = (
|
|
plausible_query(["pageviews"], MO_CUR, MAP_FILTER)[0]
|
|
> plausible_query(["pageviews"], MO_PREV, MAP_FILTER)[0]
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
map_up = None
|
|
|
|
fmt = lambda v: "?" if v is None else ("up" if v else "down") # noqa: E731
|
|
if impr_up and map_up:
|
|
verdict = "KEEP: impressions ↑ and map opens ↑ MoM"
|
|
elif impr_up is None or map_up is None:
|
|
verdict = "INCONCLUSIVE: enable GSC + Plausible to decide"
|
|
else:
|
|
verdict = "REVIEW: impressions and/or map opens did not grow MoM"
|
|
print(f" impressions MoM: {fmt(impr_up)} | map opens MoM: {fmt(map_up)}")
|
|
print(f" → {verdict}\n")
|
|
|
|
|
|
def main() -> int:
|
|
print(f"# Perfect Postcode: weekly readout ({TODAY.isoformat()})")
|
|
print(f"week {WK_CUR[0]}…{WK_CUR[1]} vs prior {WK_PREV[0]}…{WK_PREV[1]}\n")
|
|
section_seo()
|
|
section_youtube()
|
|
map_opens_weekly = section_funnel()
|
|
kill_keep(map_opens_weekly)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|