Dashboard único: coleta e alertas

1#!/usr/bin/env python3
2"""Cap. 23 - End-to-end dashboard demo with FICTIONAL data (zero credentials).
3
4Runs the whole pipeline offline:
5
6  1. Creates an in-memory SQLite DB via ``storage.py``.
7  2. Inserts several days of FICTIONAL metrics (no API, no credentials).
8  3. Evaluates the alert rules from ``alerts.py``.
9  4. Prints a small text dashboard: current snapshot, 7-day trend and alerts.
10
11Run it standalone::
12
13    python3 demo_data.py
14
15Nothing here touches Google Search Console, GA4 or any external service - the
16numbers are made up so you can see the dashboard working before wiring real
17credentials into ``collectors.py``.
18
19Part of the "living code" companion repository for the book
20"Aparecer ou Sumir" / "Be Seen or Be Forgotten".
21"""
22
23from __future__ import annotations
24
25from datetime import date, timedelta
26from typing import List
27
28import alerts
29import storage
30
31# ---------------------------------------------------------------------------
32# Fictional 7-day series. Engineered so the last day trips two alert rules:
33#   * gsc clicks fall sharply on the final day (drop_pct > 20%).
34#   * ga4 lcp creeps above the 2.5s ceiling on the final day (max).
35# ---------------------------------------------------------------------------
36CLICKS = [1180, 1205, 1240, 1260, 1230, 1275, 890]      # last day: big drop
37IMPRESSIONS = [45000, 46200, 47100, 47800, 47200, 48100, 46000]
38SESSIONS = [2900, 2980, 3050, 3110, 3020, 3140, 3090]   # stable -> no alert
39LCP = [2.0, 2.1, 2.0, 2.2, 2.3, 2.4, 2.7]               # last day: over ceiling
40SIGNUPS = [38, 41, 44, 47, 43, 49, 45]
41
42
43def _dates(n: int) -> List[str]:
44    """Return the last ``n`` dates ending today, oldest first, as 'YYYY-MM-DD'."""
45    today = date(2026, 7, 11)  # fixed for reproducible demo output
46    return [(today - timedelta(days=n - 1 - i)).isoformat() for i in range(n)]
47
48
49def seed_fictional_data(conn: storage.sqlite3.Connection) -> int:
50    """Insert the fictional multi-day series. Returns rows written."""
51    days = _dates(7)
52    rows = []
53    series = [
54        ("gsc", "clicks", CLICKS),
55        ("gsc", "impressions", IMPRESSIONS),
56        ("ga4", "sessions", SESSIONS),
57        ("ga4", "lcp", LCP),
58        ("custom", "signups", SIGNUPS),
59    ]
60    for source, metric, values in series:
61        for day, value in zip(days, values):
62            rows.append({"source": source, "metric": metric, "value": value, "date": day})
63    return storage.upsert_metrics(conn, rows)
64
65
66def _spark(values: List[float]) -> str:
67    """Tiny inline sparkline from a list of numbers (Unicode block bars)."""
68    bars = "▁▂▃▄▅▆▇█"
69    lo, hi = min(values), max(values)
70    span = (hi - lo) or 1.0
71    return "".join(bars[int((v - lo) / span * (len(bars) - 1))] for v in values)
72
73
74def print_dashboard(conn: storage.sqlite3.Connection) -> None:
75    """Render the small text dashboard from whatever is in the DB."""
76    print("=" * 60)
77    print("  DASHBOARD ÚNICO — dados fictícios (sem credenciais)")
78    print("  Unified dashboard — fictional data (no credentials)")
79    print("=" * 60)
80
81    snapshot = storage.latest_per_metric(conn)
82    print("\n  Snapshot (valor mais recente / latest value)")
83    print("  " + "-" * 46)
84    print(f"  {'source':<8}{'metric':<14}{'value':>12}  {'date':>12}")
85    for row in snapshot:
86        value = row["value"]
87        shown = f"{value:.1f}" if value != int(value) else f"{int(value)}"
88        print(f"  {row['source']:<8}{row['metric']:<14}{shown:>12}  {row['date']:>12}")
89
90    print("\n  Tendência 7 dias (7-day trend)")
91    print("  " + "-" * 46)
92    for source, metric in [
93        ("gsc", "clicks"),
94        ("gsc", "impressions"),
95        ("ga4", "sessions"),
96        ("ga4", "lcp"),
97        ("custom", "signups"),
98    ]:
99        rows = storage.query_recent(conn, metric, source=source, limit=7)
100        values = [r["value"] for r in reversed(rows)]  # oldest -> newest
101        label = f"{source}/{metric}"
102        print(f"  {label:<20} {_spark(values)}  ({values[0]:g} → {values[-1]:g})")
103
104    fired = list(alerts.evaluate(conn))
105    print("\n  Alertas (alerts)")
106    print("  " + "-" * 46)
107    if not fired:
108        print("  OK — nenhuma métrica fora do limite / no metrics off threshold.")
109    else:
110        for alert in fired:
111            print(f"  [ALERTA] {alert.message}")
112    print("=" * 60)
113
114
115def main() -> None:
116    conn = storage.connect(":memory:")
117    storage.init_db(conn)
118    written = seed_fictional_data(conn)
119    print(f"[demo] {written} linhas fictícias inseridas / fictional rows inserted\n")
120    print_dashboard(conn)
121    conn.close()
122
123
124if __name__ == "__main__":
125    main()

1#!/usr/bin/env python3
2"""Cap. 23 - Metric collectors for the unified dashboard - STRUCTURE.
3
4Defines collectors that pull data from three sources into a common shape:
5
6  * Google Search Console (Search Analytics API)
7  * Google Analytics 4 (GA4 Data API)
8  * A "custom" source (your own internal API)
9
10Each collector returns a list of metric dicts shaped like::
11
12    {"source": "gsc", "metric": "clicks", "value": 1234, "date": "2026-07-10"}
13
14so they can be fed straight into ``storage.upsert_metrics``.
15
16============================================================================
17 REQUER CREDENCIAL / REQUIRES CREDENTIALS
18 ---------------------------------------------------------------------------
19 The three real collectors below (GSC, GA4, Custom) need API credentials to
20 return live data. Credentials are read ONLY from ``os.environ`` - no secret
21 is ever hardcoded. Until they are set, the collectors raise
22 ``NotImplementedError`` with setup guidance, but the module stays importable
23 and ``py_compile``-clean. Use ``sample_rows()`` / ``collect_all(..., use_sample=True)``
24 to run the whole pipeline offline with fictional data.
25============================================================================
26
27Part of the "living code" companion repository for the book
28"Aparecer ou Sumir" / "Be Seen or Be Forgotten".
29"""
30
31from __future__ import annotations
32
33import os
34from typing import Dict, List
35
36# ---------------------------------------------------------------------------
37# TODO: credential — all values come from the environment. Nothing is hardcoded.
38# Set these before switching use_sample to False.
39# ---------------------------------------------------------------------------
40# TODO: credential — path to a Google service-account key JSON with GSC + GA4 access.
41GOOGLE_APPLICATION_CREDENTIALS = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS", "")
42# TODO: credential — the verified Search Console site URL, e.g. "https://www.example.com/".
43GSC_SITE_URL = os.environ.get("GSC_SITE_URL", "")
44# TODO: credential — the GA4 numeric property id, e.g. "properties/123456789".
45GA4_PROPERTY_ID = os.environ.get("GA4_PROPERTY_ID", "")
46# TODO: credential — base URL + token for your own custom metrics API.
47CUSTOM_API_BASE = os.environ.get("CUSTOM_API_BASE", "")
48CUSTOM_API_TOKEN = os.environ.get("CUSTOM_API_TOKEN", "")
49
50
51MetricRow = Dict[str, object]
52
53
54class Collector:
55    """Base collector. Subclasses implement :meth:`collect`."""
56
57    source = "base"
58
59    def collect(self, date: str) -> List[MetricRow]:
60        """Return metric rows for the given 'YYYY-MM-DD' date."""
61        raise NotImplementedError
62
63
64class SearchConsoleCollector(Collector):
65    """Collector for the Google Search Console Search Analytics API.
66
67    REQUIRES CREDENTIALS: GOOGLE_APPLICATION_CREDENTIALS + GSC_SITE_URL.
68    """
69
70    source = "gsc"
71
72    def collect(self, date: str) -> List[MetricRow]:
73        """Pull clicks/impressions/CTR/position for ``date`` from GSC.
74
75        TODO: credential — steps once you have a service account:
76          1. pip install google-api-python-client google-auth
77          2. Build the service with GOOGLE_APPLICATION_CREDENTIALS::
78
79                 from googleapiclient.discovery import build
80                 from google.oauth2.service_account import Credentials
81                 creds = Credentials.from_service_account_file(
82                     GOOGLE_APPLICATION_CREDENTIALS,
83                     scopes=["https://www.googleapis.com/auth/webmasters.readonly"])
84                 service = build("searchconsole", "v1", credentials=creds)
85          3. Call ``searchanalytics.query`` on GSC_SITE_URL for the date range.
86        """
87        if not (GOOGLE_APPLICATION_CREDENTIALS and GSC_SITE_URL):
88            raise NotImplementedError(
89                "SearchConsoleCollector needs GOOGLE_APPLICATION_CREDENTIALS and "
90                "GSC_SITE_URL. See the docstring for the setup steps."
91            )
92        # TODO: credential — replace with the real searchanalytics.query call.
93        raise NotImplementedError("GSC API call not implemented yet.")
94
95
96class GA4Collector(Collector):
97    """Collector for the Google Analytics 4 Data API.
98
99    REQUIRES CREDENTIALS: GOOGLE_APPLICATION_CREDENTIALS + GA4_PROPERTY_ID.
100    """
101
102    source = "ga4"
103
104    def collect(self, date: str) -> List[MetricRow]:
105        """Pull sessions/engagement/conversions for ``date`` from GA4.
106
107        TODO: credential — steps once you have a service account:
108          1. pip install google-analytics-data
109          2. Authenticate with GOOGLE_APPLICATION_CREDENTIALS and run a
110             ``runReport`` request against GA4_PROPERTY_ID with the metrics and
111             the date range you need.
112        """
113        if not (GOOGLE_APPLICATION_CREDENTIALS and GA4_PROPERTY_ID):
114            raise NotImplementedError(
115                "GA4Collector needs GOOGLE_APPLICATION_CREDENTIALS and "
116                "GA4_PROPERTY_ID. See the docstring for the setup steps."
117            )
118        # TODO: credential — replace with the real runReport call.
119        raise NotImplementedError("GA4 API call not implemented yet.")
120
121
122class CustomCollector(Collector):
123    """Collector for your own internal metrics API.
124
125    REQUIRES CREDENTIALS: CUSTOM_API_BASE + CUSTOM_API_TOKEN.
126    """
127
128    source = "custom"
129
130    def collect(self, date: str) -> List[MetricRow]:
131        """Pull custom metrics for ``date`` from your own API.
132
133        TODO: credential — perform an authenticated HTTP request to
134        CUSTOM_API_BASE using CUSTOM_API_TOKEN and map the response into
135        MetricRow dicts.
136        """
137        if not (CUSTOM_API_BASE and CUSTOM_API_TOKEN):
138            raise NotImplementedError(
139                "CustomCollector needs CUSTOM_API_BASE and CUSTOM_API_TOKEN."
140            )
141        # TODO: credential — replace with the real HTTP request.
142        raise NotImplementedError("Custom API call not implemented yet.")
143
144
145def sample_rows(date: str) -> List[MetricRow]:
146    """Return sample-shaped rows so the pipeline can be exercised offline.
147
148    Useful for local development, tests and demos - it mimics exactly what the
149    real collectors would return, without touching any external service or
150    needing any credential.
151    """
152    return [
153        {"source": "gsc", "metric": "clicks", "value": 1250, "date": date},
154        {"source": "gsc", "metric": "impressions", "value": 48000, "date": date},
155        {"source": "ga4", "metric": "sessions", "value": 3100, "date": date},
156        {"source": "ga4", "metric": "lcp", "value": 2.1, "date": date},
157        {"source": "custom", "metric": "signups", "value": 42, "date": date},
158    ]
159
160
161def collect_all(date: str, use_sample: bool = True) -> List[MetricRow]:
162    """Collect from every source for ``date``.
163
164    When ``use_sample`` is True (default) it returns offline sample data so the
165    dashboard can run without credentials. Set it to False once the collectors
166    above are implemented AND their credentials are configured to hit the real
167    APIs.
168    """
169    if use_sample:
170        return sample_rows(date)
171
172    rows: List[MetricRow] = []
173    for collector in (SearchConsoleCollector(), GA4Collector(), CustomCollector()):
174        rows.extend(collector.collect(date))
175    return rows
176
177
178if __name__ == "__main__":
179    # Demo without credentials: print the sample rows.
180    for r in collect_all("2026-07-10"):
181        print(r)

1#!/usr/bin/env python3
2"""Cap. 23 - SQLite storage layer for the unified dashboard.
3
4A tiny, fully functional persistence layer built on the stdlib ``sqlite3``.
5Every metric collected from Google Search Console, GA4 or a custom source is
6stored in a single ``metrics`` table so the dashboard and alert rules can query
7them uniformly. No external dependencies, no credentials.
8
9Part of the "living code" companion repository for the book
10"Aparecer ou Sumir" / "Be Seen or Be Forgotten".
11"""
12
13from __future__ import annotations
14
15import sqlite3
16from pathlib import Path
17from typing import Iterable, List, Mapping, Optional, Union
18
19# Default on-disk database file for the dashboard.
20DEFAULT_DB_PATH = "dashboard.db"
21
22
23def connect(db_path: Union[str, Path] = DEFAULT_DB_PATH) -> sqlite3.Connection:
24    """Open a SQLite connection with row access by column name."""
25    conn = sqlite3.connect(str(db_path))
26    conn.row_factory = sqlite3.Row
27    return conn
28
29
30def init_db(conn: sqlite3.Connection) -> None:
31    """Create the metrics table (and a helpful unique index) if missing.
32
33    Schema:
34      source  - where the metric came from (e.g. 'gsc', 'ga4', 'custom')
35      metric  - the metric name (e.g. 'clicks', 'lcp', 'signups')
36      value   - the numeric value
37      date    - the day the metric refers to, as 'YYYY-MM-DD'
38
39    The UNIQUE(source, metric, date) index makes upserts idempotent: re-running a
40    collector for the same day updates the value instead of duplicating rows.
41    """
42    conn.execute(
43        """
44        CREATE TABLE IF NOT EXISTS metrics (
45            id      INTEGER PRIMARY KEY AUTOINCREMENT,
46            source  TEXT    NOT NULL,
47            metric  TEXT    NOT NULL,
48            value   REAL    NOT NULL,
49            date    TEXT    NOT NULL
50        )
51        """
52    )
53    conn.execute(
54        """
55        CREATE UNIQUE INDEX IF NOT EXISTS ux_metrics_source_metric_date
56        ON metrics (source, metric, date)
57        """
58    )
59    conn.commit()
60
61
62def upsert_metric(
63    conn: sqlite3.Connection,
64    source: str,
65    metric: str,
66    value: float,
67    date: str,
68) -> None:
69    """Insert or update a single metric row keyed by (source, metric, date)."""
70    conn.execute(
71        """
72        INSERT INTO metrics (source, metric, value, date)
73        VALUES (?, ?, ?, ?)
74        ON CONFLICT(source, metric, date) DO UPDATE SET value = excluded.value
75        """,
76        (source, metric, value, date),
77    )
78    conn.commit()
79
80
81def upsert_metrics(conn: sqlite3.Connection, rows: Iterable[Mapping[str, object]]) -> int:
82    """Bulk upsert. Each row is a mapping with source/metric/value/date keys.
83
84    Returns the number of rows processed.
85    """
86    prepared = [
87        (r["source"], r["metric"], float(r["value"]), r["date"]) for r in rows
88    ]
89    conn.executemany(
90        """
91        INSERT INTO metrics (source, metric, value, date)
92        VALUES (?, ?, ?, ?)
93        ON CONFLICT(source, metric, date) DO UPDATE SET value = excluded.value
94        """,
95        prepared,
96    )
97    conn.commit()
98    return len(prepared)
99
100
101def query_recent(
102    conn: sqlite3.Connection,
103    metric: str,
104    source: Optional[str] = None,
105    limit: int = 30,
106) -> List[sqlite3.Row]:
107    """Return the most recent rows for a metric, newest first.
108
109    Args:
110        metric: metric name to filter on (e.g. 'clicks').
111        source: optional source filter (e.g. 'gsc').
112        limit: maximum number of rows to return.
113    """
114    if source is None:
115        cur = conn.execute(
116            "SELECT source, metric, value, date FROM metrics "
117            "WHERE metric = ? ORDER BY date DESC LIMIT ?",
118            (metric, limit),
119        )
120    else:
121        cur = conn.execute(
122            "SELECT source, metric, value, date FROM metrics "
123            "WHERE metric = ? AND source = ? ORDER BY date DESC LIMIT ?",
124            (metric, source, limit),
125        )
126    return cur.fetchall()
127
128
129def latest_per_metric(conn: sqlite3.Connection) -> List[sqlite3.Row]:
130    """Return the single newest row for every (source, metric) pair.
131
132    Handy for rendering a "current state" dashboard snapshot.
133    """
134    cur = conn.execute(
135        """
136        SELECT m.source, m.metric, m.value, m.date
137        FROM metrics AS m
138        JOIN (
139            SELECT source, metric, MAX(date) AS max_date
140            FROM metrics
141            GROUP BY source, metric
142        ) AS latest
143        ON m.source = latest.source
144        AND m.metric = latest.metric
145        AND m.date = latest.max_date
146        ORDER BY m.source, m.metric
147        """
148    )
149    return cur.fetchall()
150
151
152if __name__ == "__main__":
153    # Tiny self-check that runs without any external credentials.
154    connection = connect(":memory:")
155    init_db(connection)
156    upsert_metric(connection, "gsc", "clicks", 1200, "2026-07-10")
157    upsert_metric(connection, "gsc", "clicks", 1300, "2026-07-11")
158    for row in query_recent(connection, "clicks"):
159        print(dict(row))
160    connection.close()

1#!/usr/bin/env python3
2"""Cap. 23 - Threshold-based alerts for the unified dashboard.
3
4Reads recent metrics from the SQLite storage layer and yields alerts when a
5threshold is crossed. Two kinds of rules are demonstrated:
6
7  * DROP rules: fire when a metric drops more than X% versus the previous value
8    (e.g. organic clicks fall > 20% day over day).
9  * MAX rules: fire when a metric exceeds an absolute ceiling
10    (e.g. Largest Contentful Paint > 2.5s).
11
12Pure Python, no external services, no credentials. The ``__main__`` demo inserts
13sample data via ``storage.py`` so it runs standalone.
14
15Part of the "living code" companion repository for the book
16"Aparecer ou Sumir" / "Be Seen or Be Forgotten".
17"""
18
19from __future__ import annotations
20
21from dataclasses import dataclass
22from typing import Iterator, List, Optional
23
24import storage
25
26
27@dataclass
28class Threshold:
29    """A single alert rule.
30
31    kind:
32      'drop_pct' - fire when value falls more than ``limit`` percent vs previous.
33      'max'      - fire when the latest value is greater than ``limit``.
34    """
35
36    metric: str
37    kind: str
38    limit: float
39    source: Optional[str] = None  # optional source filter
40
41
42# Example threshold configuration. Tune these to your own baselines.
43EXAMPLE_THRESHOLDS: List[Threshold] = [
44    Threshold(metric="clicks", kind="drop_pct", limit=20.0, source="gsc"),
45    Threshold(metric="sessions", kind="drop_pct", limit=25.0, source="ga4"),
46    Threshold(metric="lcp", kind="max", limit=2.5, source="ga4"),
47]
48
49
50@dataclass
51class Alert:
52    """A fired alert, ready to be logged, emailed or posted to Slack."""
53
54    metric: str
55    source: str
56    message: str
57    value: float
58
59
60def evaluate(
61    conn: "storage.sqlite3.Connection",
62    thresholds: Optional[List[Threshold]] = None,
63) -> Iterator[Alert]:
64    """Yield an Alert for every threshold that is currently crossed.
65
66    For each rule we pull the two most recent stored values for the metric and
67    compare them (drop rules) or check the latest against a ceiling (max rules).
68    """
69    rules = thresholds if thresholds is not None else EXAMPLE_THRESHOLDS
70
71    for rule in rules:
72        rows = storage.query_recent(conn, rule.metric, source=rule.source, limit=2)
73        if not rows:
74            continue
75        latest = rows[0]["value"]
76        src = rows[0]["source"]
77
78        if rule.kind == "max":
79            if latest > rule.limit:
80                yield Alert(
81                    metric=rule.metric,
82                    source=src,
83                    value=latest,
84                    message=(
85                        f"{rule.metric} = {latest} exceeds max {rule.limit} "
86                        f"(source: {src})"
87                    ),
88                )
89
90        elif rule.kind == "drop_pct":
91            if len(rows) < 2:
92                continue  # need a previous value to compute a drop
93            previous = rows[1]["value"]
94            if previous <= 0:
95                continue
96            drop_pct = (previous - latest) / previous * 100.0
97            if drop_pct > rule.limit:
98                yield Alert(
99                    metric=rule.metric,
100                    source=src,
101                    value=latest,
102                    message=(
103                        f"{rule.metric} dropped {drop_pct:.1f}% "
104                        f"({previous} -> {latest}) exceeding {rule.limit}% "
105                        f"(source: {src})"
106                    ),
107                )
108
109
110def _demo() -> None:
111    """Standalone demo: build an in-memory DB, insert sample data, run alerts."""
112    conn = storage.connect(":memory:")
113    storage.init_db(conn)
114
115    # Sample data engineered to trigger both a drop alert and a max alert.
116    storage.upsert_metrics(
117        conn,
118        [
119            # clicks fell from 1500 -> 1000 (~33% drop) -> triggers drop_pct.
120            {"source": "gsc", "metric": "clicks", "value": 1500, "date": "2026-07-10"},
121            {"source": "gsc", "metric": "clicks", "value": 1000, "date": "2026-07-11"},
122            # sessions roughly stable -> no alert.
123            {"source": "ga4", "metric": "sessions", "value": 3000, "date": "2026-07-10"},
124            {"source": "ga4", "metric": "sessions", "value": 2950, "date": "2026-07-11"},
125            # lcp = 3.2s > 2.5s ceiling -> triggers max.
126            {"source": "ga4", "metric": "lcp", "value": 3.2, "date": "2026-07-11"},
127        ],
128    )
129
130    alerts = list(evaluate(conn))
131    if not alerts:
132        print("No alerts. All metrics within thresholds.")
133    for alert in alerts:
134        print(f"[ALERT] {alert.message}")
135    conn.close()
136
137
138if __name__ == "__main__":
139    _demo()