#!/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())