Implementation Notes: Building a Personal Blog Operations Dashboard with Streamlit and the Gemini API

# auth.py — shared auth helpers
from google.oauth2 import service_account
from googleapiclient.discovery import build
from google.analytics.data_v1beta import BetaAnalyticsDataClient
 
SCOPES = [
    "https://www.googleapis.com/auth/webmasters.readonly",
    "https://www.googleapis.com/auth/analytics.readonly",
]
 
def get_creds(key_path: str = "service_account.json"):
    return service_account.Credentials.from_service_account_file(
        key_path, scopes=SCOPES
    )
 
def gsc_client(creds):
    return build("searchconsole", "v1", credentials=creds)
 
def ga4_client(creds):
    return BetaAnalyticsDataClient(credentials=creds)

# .streamlit/secrets.toml
[service_account]
type = "service_account"
project_id = "your-project"
private_key_id = "..."
private_key = "-----BEGIN PRIVATE KEY-----\n..."
client_email = "blog-dashboard@your-project.iam.gserviceaccount.com"

# gsc.py
import pandas as pd
from datetime import date, timedelta
 
def fetch_gsc(client, site_url: str, days: int = 28) -> dict:
    end = date.today() - timedelta(days=2)  # GSC lags ~2 days
    start = end - timedelta(days=days)
    
    body_query = {
        "startDate": start.isoformat(),
        "endDate": end.isoformat(),
        "dimensions": ["query"],
        "rowLimit": 5000,
    }
    body_page = {**body_query, "dimensions": ["page"]}
    
    queries = client.searchanalytics().query(
        siteUrl=site_url, body=body_query
    ).execute().get("rows", [])
    
    pages = client.searchanalytics().query(
        siteUrl=site_url, body=body_page
    ).execute().get("rows", [])
    
    df_q = pd.DataFrame([{
        "query": r["keys"][0],
        "clicks": r["clicks"],
        "impressions": r["impressions"],
        "ctr": r["ctr"],
        "position": r["position"],
    } for r in queries])
    
    df_p = pd.DataFrame([{
        "page": r["keys"][0],
        "clicks": r["clicks"],
        "impressions": r["impressions"],
        "ctr": r["ctr"],
        "position": r["position"],
    } for r in pages])
    
    return {"queries": df_q, "pages": df_p}

# ga4.py
from google.analytics.data_v1beta.types import (
    DateRange, Dimension, Metric, RunReportRequest,
)
 
def fetch_ga4(client, property_id: str, days: int = 28) -> pd.DataFrame:
    request = RunReportRequest(
        property=f"properties/{property_id}",
        dimensions=[Dimension(name="pagePath")],
        metrics=[
            Metric(name="activeUsers"),
            Metric(name="screenPageViews"),
            Metric(name="averageSessionDuration"),
            Metric(name="engagementRate"),
        ],
        date_ranges=[DateRange(
            start_date=f"{days}daysAgo", end_date="today"
        )],
        limit=10000,
    )
    response = client.run_report(request)
    
    return pd.DataFrame([{
        "page": row.dimension_values[0].value,
        "active_users": int(row.metric_values[0].value),
        "page_views": int(row.metric_values[1].value),
        "avg_duration": float(row.metric_values[2].value),
        "engagement_rate": float(row.metric_values[3].value),
    } for row in response.rows])

def normalize_page(url: str, site_origin: str) -> str:
    if url.startswith("http"):
        return url.replace(site_origin, "")
    return url
 
gsc["page"] = gsc["page"].apply(
    lambda u: normalize_page(u, "https://gemilab.net")
)
merged = pd.merge(gsc, ga4, on="page", how="outer")

# quality.py
from google import genai
from google.genai import types
 
PROMPT = """You are a quality reviewer well-versed in Google Search Console and
the Helpful Content System. Read the article and score quality from 0 to 100.
 
Axes:
- Originality (is this just a doc summary?): 30 pts
- Practicality (can the reader take action after reading?): 30 pts
- Tone (is it falling into templated AI explanation voice?): 20 pts
- Structure (H2 organization, code examples, metrics balance): 20 pts
 
Output JSON only:
{
  "score": 75,
  "uniqueness": 25,
  "practicality": 22,
  "tone": 16,
  "structure": 12,
  "deletion_candidate": false,
  "improvements": ["intro reads templated", "code lacks explanation"]
}
 
Set deletion_candidate to true if score < 60.
 
---
 
{article_body}
"""
 
def score_article(client, article_body: str) -> dict:
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=PROMPT.format(article_body=article_body[:30000]),
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            temperature=0.1,
        ),
    )
    import json
    return json.loads(response.text)

# app.py
import streamlit as st
import pandas as pd
from auth import get_creds, gsc_client, ga4_client
from gsc import fetch_gsc
from ga4 import fetch_ga4
 
SITES = {
    "Claude Lab": {
        "url": "sc-domain:claudelab.net",
        "ga4": "123456789",
        "origin": "https://claudelab.net",
    },
    "Gemini Lab": {
        "url": "sc-domain:gemilab.net",
        "ga4": "234567890",
        "origin": "https://gemilab.net",
    },
}
 
@st.cache_data(ttl=3600)
def load_gsc(site_url, days):
    return fetch_gsc(gsc_client(get_creds()), site_url, days)
 
@st.cache_data(ttl=3600)
def load_ga4(property_id, days):
    return fetch_ga4(ga4_client(get_creds()), property_id, days)
 
st.sidebar.title("Operations Dashboard")
site_key = st.sidebar.selectbox("Site", list(SITES.keys()))
days = st.sidebar.slider("Days", 7, 90, 28)
site = SITES[site_key]
 
gsc = load_gsc(site["url"], days)
ga4 = load_ga4(site["ga4"], days)
 
totals = gsc["queries"]
col1, col2, col3, col4 = st.columns(4)
col1.metric("Clicks", f"{totals['clicks'].sum():,}")
col2.metric("Impressions", f"{totals['impressions'].sum():,}")
col3.metric("CTR", f"{totals['ctr'].mean()*100:.1f}%")
col4.metric("Avg Position", f"{totals['position'].mean():.1f}")
 
st.subheader("Top queries (by clicks)")
st.dataframe(
    totals.sort_values("clicks", ascending=False).head(30),
    use_container_width=True,
)

# weekly_scan.py
import os, json
from pathlib import Path
from quality import score_article
from google import genai
 
def scan_site(site_dir: Path, output: Path):
    client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
    candidates = []
    
    for mdx in site_dir.rglob("*.mdx"):
        body = mdx.read_text(encoding="utf-8")
        try:
            score = score_article(client, body)
        except Exception as e:
            print(f"skip {mdx}: {e}")
            continue
        
        if score["deletion_candidate"]:
            candidates.append({
                "path": str(mdx.relative_to(site_dir)),
                "score": score["score"],
                "improvements": score["improvements"],
            })
    
    output.write_text(json.dumps(candidates, ensure_ascii=False, indent=2))
    print(f"{len(candidates)} candidates -> {output}")
 
if __name__ == "__main__":
    for name, dir in [
        ("claudelab", Path("/tmp/repos/claudelab.net/content/articles/ja")),
        ("gemilab", Path("/tmp/repos/gemilab.net/content/articles/ja")),
    ]:
        scan_site(dir, Path(f"/tmp/scan_{name}.json"))