Serverless Analytics Warehouse with BigQuery & Cloud Run: From GA4 Streams to Automated SEO Alerts

┌─────────────────┐       ┌─────────────────┐       ┌─────────────────────┐
│  Web / App      │       │ Google Analytics│       │ BigQuery Raw Stream │
│  (Next.js Blog) ├──────►│ 4 (GA4 Events)  ├──────►│ `analytics_xxxxxxx` │
└─────────────────┘       └─────────────────┘       │ (Partitioned Tables)│
                                                    └──────────┬──────────┘
                                                               │
                                                 Daily Scheduled Query / View
                                                               │
                                                               ▼
┌─────────────────┐       ┌─────────────────┐       ┌─────────────────────┐
│  Telegram /     │       │ Cloud Run Job   │       │ Curated Mart:       │
│  Slack Alert    │◄──────┤ (Fast Python    │◄──────┤ `daily_seo_dropoffs`│
│  (Actionable)   │       │  Anomaly Worker)│       │ (Summary Table)     │
└─────────────────┘       └─────────────────┘       └─────────────────────┘

-- Create or replace curated SEO analytics table
CREATE OR REPLACE TABLE `locionic_analytics.daily_post_performance`
PARTITION BY event_date
CLUSTER BY post_slug, locale AS

WITH raw_events AS (
  SELECT
    PARSE_DATE('%Y%m%d', event_date) AS event_date,
    (SELECT value.string_value FROM UNNEST(event_params) WHERE key = 'page_location') AS full_url,
    (SELECT value.int_value FROM UNNEST(event_params) WHERE key = 'ga_session_id') AS session_id,
    user_pseudo_id,
    event_name
  FROM
    `project-cf6c933d-0d23-4aae-af7.analytics_324892182.events_*`
  WHERE
    _TABLE_SUFFIX = FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY))
),

parsed_pages AS (
  SELECT
    event_date,
    user_pseudo_id,
    session_id,
    REGEXP_EXTRACT(full_url, r'https?://[^/]+/([^/]+)/blog/([^/?#]+)') AS locale,
    REGEXP_EXTRACT(full_url, r'https?://[^/]+/[^/]+/blog/([^/?#]+)') AS post_slug
  FROM
    raw_events
  WHERE
    event_name = 'page_view'
    AND full_url LIKE '%/blog/%'
)

SELECT
  event_date,
  post_slug,
  COALESCE(locale, 'en') AS locale,
  COUNT(1) AS total_views,
  COUNT(DISTINCT user_pseudo_id) AS unique_readers,
  COUNT(DISTINCT session_id) AS total_sessions
FROM
  parsed_pages
WHERE
  post_slug IS NOT NULL
GROUP BY
  event_date,
  post_slug,
  locale;

import os
import json
import requests
from google.cloud import bigquery

PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "project-cf6c933d-0d23-4aae-af7")
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID")

def run_anomaly_check():
    client = bigquery.Client(project=PROJECT_ID)
    
    # Identify pages where views dropped by >40% compared to 7-day average
    query = """
    WITH seven_day_stats AS (
      SELECT
        post_slug,
        AVG(total_views) AS avg_views_7d
      FROM
        `locionic_analytics.daily_post_performance`
      WHERE
        event_date BETWEEN DATE_SUB(CURRENT_DATE(), INTERVAL 8 DAY) AND DATE_SUB(CURRENT_DATE(), INTERVAL 2 DAY)
      GROUP BY post_slug
      HAVING avg_views_7d >= 10
    ),
    yesterday_stats AS (
      SELECT
        post_slug,
        total_views AS views_yesterday
      FROM
        `locionic_analytics.daily_post_performance`
      WHERE
        event_date = DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY)
    )
    SELECT
      s.post_slug,
      ROUND(s.avg_views_7d, 1) AS avg_7d,
      COALESCE(y.views_yesterday, 0) AS yesterday,
      ROUND(((COALESCE(y.views_yesterday, 0) - s.avg_views_7d) / s.avg_views_7d) * 100, 1) AS pct_change
    FROM
      seven_day_stats s
    LEFT JOIN
      yesterday_stats y ON s.post_slug = y.post_slug
    WHERE
      COALESCE(y.views_yesterday, 0) < (s.avg_views_7d * 0.6)
    ORDER BY
      pct_change ASC;
    """
    
    query_job = client.query(query)
    results = list(query_job.result())
    
    if not results:
        print("No traffic anomalies detected today.")
        return

    lines = ["⚠️ *Locionic SEO Alert: Traffic Drop Detected*"]
    for row in results:
        lines.append(f"• `{row.post_slug}`: {row.yesterday} views vs {row.avg_7d} avg ({row.pct_change}%)")

    message = "\n".join(lines)
    print(message)
    
    if TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID:
        requests.post(
            f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
            json={"chat_id": TELEGRAM_CHAT_ID, "text": message, "parse_mode": "Markdown"},
            timeout=10
        )

if __name__ == "__main__":
    run_anomaly_check()