"""Generates a self-contained HTML dashboard from pipeline data.
The dashboard is a single static file with data baked in as a JSON blob and
client-side rendering via Chart.js. It's intended to be regenerated by the
weekly cron alongside the Sheets writes — open the file in any browser.
"""
import json
from collections import defaultdict
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Optional
from zoneinfo import ZoneInfo
from models import Person
def write_dashboard(people: list[Person], config: dict, tz: ZoneInfo) -> str:
"""Render the dashboard HTML and write it to dashboard.output_path. Returns the path."""
dash_config = config.get("dashboard", {}) or {}
output_path = dash_config.get("output_path", "dashboard.html")
deal_label = dash_config.get("deal_label", "deal")
deal_launch = _parse_launch_date(dash_config.get("deal_launch_date"), tz)
data = _build_data(people, deal_launch, deal_label, tz)
html = _render(data)
Path(output_path).write_text(html, encoding="utf-8")
return output_path
def _parse_launch_date(value: Any, tz: ZoneInfo) -> Optional[datetime]:
if not value:
return None
if isinstance(value, datetime):
return value if value.tzinfo else value.replace(tzinfo=tz)
if isinstance(value, date):
return datetime(value.year, value.month, value.day, tzinfo=tz)
if isinstance(value, str):
return datetime.fromisoformat(value).replace(tzinfo=tz)
return None
def _build_data(
people: list[Person],
deal_launch: Optional[datetime],
deal_label: str,
tz: ZoneInfo,
) -> dict[str, Any]:
now = datetime.now(timezone.utc)
has_newsletter = sum(1 for p in people if p.newsletter_subscribed_at)
has_newspaper = sum(1 for p in people if p.newspaper_subscribed_at)
both = sum(
1 for p in people if p.newsletter_subscribed_at and p.newspaper_subscribed_at
)
active_newspaper = sum(
1 for p in people if p.newspaper_subscription_status == "active"
)
nl_to_np_rate = (100 * both / has_newsletter) if has_newsletter else 0.0
deal_people_all = [p for p in people if p.deal_started_at is not None]
if deal_launch:
deal_cohort = [
p for p in deal_people_all if p.deal_started_at >= deal_launch
]
pre_launch = [
p for p in deal_people_all if p.deal_started_at < deal_launch
]
else:
deal_cohort = deal_people_all
pre_launch = []
in_trial = [p for p in deal_cohort if p.deal_outcome == "in_trial"]
converted = [p for p in deal_cohort if p.deal_outcome == "converted"]
lost = [p for p in deal_cohort if p.deal_outcome == "lost"]
completed = [
p for p in deal_cohort if p.deal_trial_end and p.deal_trial_end < now
]
completed_converted = [p for p in completed if p.deal_outcome == "converted"]
trial_conversion_rate = (
100 * len(completed_converted) / len(completed) if completed else 0.0
)
today_local = datetime.now(tz).date()
end_week = _iso_week_start(today_local)
start_week = end_week - timedelta(weeks=52)
weekly: dict[date, dict[str, int]] = defaultdict(
lambda: {"newsletter": 0, "newspaper": 0, "deal": 0}
)
for p in people:
for field, key in (
(p.newsletter_subscribed_at, "newsletter"),
(p.newspaper_subscribed_at, "newspaper"),
(p.deal_started_at, "deal"),
):
if field is None:
continue
wk = _iso_week_start(field.astimezone(tz).date())
if start_week <= wk <= end_week:
weekly[wk][key] += 1
weeks_data = []
cur = start_week
while cur <= end_week:
d = weekly[cur]
weeks_data.append(
{
"week": cur.isoformat(),
"newsletter": d["newsletter"],
"newspaper": d["newspaper"],
"deal": d["deal"],
}
)
cur += timedelta(weeks=1)
deal_launch_week = (
_iso_week_start(deal_launch.astimezone(tz).date()).isoformat()
if deal_launch
else None
)
table_rows = []
for p in people:
table_rows.append(
{
"e": p.email,
"ft": _iso(p.first_touch_at),
"fs": p.first_touch_source,
"nl": _iso(p.newsletter_subscribed_at),
"np": _iso(p.newspaper_subscribed_at),
"ns": p.newspaper_subscription_status or "",
"dn": (
round(p.days_to_newspaper, 1)
if p.days_to_newspaper is not None
else None
),
"ds": _iso(p.deal_started_at),
"do": p.deal_outcome or "",
}
)
return {
"generated_at": now.astimezone(tz).strftime("%Y-%m-%d %H:%M:%S %Z"),
"deal_label": deal_label,
"deal_launch": deal_launch.strftime("%Y-%m-%d") if deal_launch else None,
"deal_launch_week": deal_launch_week,
"kpis": {
"newsletter": has_newsletter,
"newspaper": has_newspaper,
"active_newspaper": active_newspaper,
"newsletter_to_newspaper_rate": round(nl_to_np_rate, 1),
"deal_total": len(deal_cohort),
"deal_in_trial": len(in_trial),
"deal_converted": len(converted),
"deal_lost": len(lost),
"deal_trial_conversion_rate": round(trial_conversion_rate, 1),
"deal_completed_trials": len(completed),
"deal_pre_launch": len(pre_launch),
},
"weekly": weeks_data,
"people": table_rows,
}
def _iso_week_start(d: date) -> date:
return d - timedelta(days=d.weekday())
def _iso(dt: Optional[datetime]) -> Optional[str]:
return dt.isoformat() if dt else None
def _render(data: dict) -> str:
payload = json.dumps(data, default=str).replace("
Funnel Tracker - Lewiston Tribune