"""Streamlit dashboard for feed status (Monty POC)."""
from datetime import date as date_cls
from urllib.parse import quote
import streamlit as st
from feed_status.ui.data import (
FEED_COL,
FEED_ID_COL,
FRESH_COL,
STABLE_COL,
fetch_section,
fetch_service_links,
get_links,
)
from feed_status.ui.sections import SECTIONS
from feed_status.ui.status import LEGENDS
st.set_page_config(page_title='Monty', layout='wide')
_NAV_CSS = """
"""
_NAV_LINKS = (
('Fact Analytics', '#fact-analytics', False),
('YT TheOrchard', '#yt-theorchard', False),
('YT SME', '#yt-sme', False),
('Market Share', '#marketshare', False),
('Monthly', '#monthly', False),
('Sender Status', '#outgoing', False),
('Activity Detector', '#activity_detector', False),
('dbt docs', '/dbt/index.html', True),
('Delphi', '/delphi', True),
('Looker', 'https://theorchard.looker.com/dashboards/1314', True),
)
def _link_html(label, href, external):
"""Render one nav link as an anchor tag."""
cls = ' class="external"' if external else ''
target = ' target="_blank"' if external else ''
return f'{label}'
def _render_nav():
"""Render the top navigation bar (anchors + external links)."""
links = ''.join(_link_html(*item) for item in _NAV_LINKS)
st.markdown(
f'{_NAV_CSS}
'
f'Monty 3.0{links}
',
unsafe_allow_html=True,
)
def _swf_run_url(day):
"""Build the AWS SWF console URL for the run."""
domain = day.get('swf_domain')
workflow_id = day.get('swf_workflow_id')
run_id = day.get('swf_run_id')
if not domain or not workflow_id or not run_id:
return None
region = 'us-east-1'
return (
f'https://{region}.console.aws.amazon.com/swf/v2/home'
f'?region={region}#/domains/{domain}/executions/'
f'{quote(workflow_id, safe="")}/{quote(run_id, safe="")}'
)
@st.dialog('Cell detail', width='large')
def _cell_dialog(data):
"""Modal dialog showing details for the clicked cell."""
title = data['feed_name']
if data['date_col']:
title += f' · {data["date_col"]}'
st.markdown(f'### {title}')
if data['date_col']:
row1 = st.columns(2)
row1[0].metric('Status', data['raw_status'] or '—')
row1[1].metric(
'Days back',
data['days_back'] if data['days_back'] is not None else '—',
)
row2 = st.columns(2)
row2[0].metric('Updated (UTC)', data['updated_at'] or '—')
row2[1].metric('Complete status', data['complete_status'] or '—')
swf_url = data.get('swf_run_url')
if swf_url:
st.markdown(f'[SWF RUN ↗]({swf_url})')
st.divider()
links = data.get('links') or {}
if links:
st.markdown('**Service links**')
st.markdown(
' · '.join(f'[{name}]({url})' for name, url in links.items()),
)
else:
st.caption('No service links configured for this feed.')
def render_section(section, date_arg):
"""Render filters, data table, legend, and the cell-detail dialog hook."""
try:
df, dates, details = fetch_section(section.apipath, date_arg)
except Exception as exc:
st.error(f'Failed to load {section.title}: {exc}')
return
col_search, col_errors = st.columns([1, 3], vertical_alignment='bottom')
with col_search:
query = st.text_input(
'Search',
key=f'search-{section.id}',
placeholder='Search feeds…',
label_visibility='collapsed',
)
with col_errors:
errors_only = st.checkbox('Errors only', key=f'errors-{section.id}')
view = df
if errors_only:
view = view[
view[FRESH_COL].str.contains('critical', na=False)
| view[STABLE_COL].str.contains('critical', na=False)
]
if query:
view = view[view[FEED_COL].str.contains(query, case=False, na=False)]
filter_token = f'{query}|{int(errors_only)}'
date_cols = {
d: st.column_config.TextColumn(
label='/'.join(reversed(d.split('-')[1:])),
width=30
)
for d in dates
}
table_height = min(max((len(view) + 1) * 35 + 3, 100), 800)
event = st.dataframe(
view,
hide_index=True,
width='stretch',
height=table_height,
on_select='rerun',
selection_mode='single-cell',
key=f'table-{section.id}-{filter_token}',
column_order=(FEED_COL, FRESH_COL, STABLE_COL, *dates),
column_config={
FEED_COL: st.column_config.TextColumn(
FEED_COL, width='medium', pinned=True,
),
FRESH_COL: st.column_config.TextColumn(FRESH_COL, width='small'),
STABLE_COL: st.column_config.TextColumn(STABLE_COL, width='small'),
**date_cols,
},
)
legend = LEGENDS.get(section.kind, [])
if legend:
st.caption(' · '.join(f'{glyph} {label}' for glyph, label in legend))
_open_dialog_if_new_selection(view, dates, details, event, section.id)
def _open_dialog_if_new_selection(view, dates, details, event, section_id):
"""Open the cell-detail dialog when the selected cell has just changed."""
if not event:
return
selection = event.get('selection') or {}
cells = selection.get('cells') or []
if not cells:
return
row_idx, column = cells[0]
cell_key = (row_idx, column)
last_key = f'_last_shown_cell_{section_id}'
if st.session_state.get(last_key) == cell_key:
return
st.session_state[last_key] = cell_key
try:
record = view.iloc[row_idx]
except IndexError:
return
feed_id = record[FEED_ID_COL]
feed_name = record[FEED_COL]
date_col = column if column in dates else None
feed_details = details.get(feed_id, {})
day = (feed_details.get('status_details') or {}).get(date_col or '', {})
raw_status = (feed_details.get('statuses') or {}).get(date_col or '')
try:
supported, links_cfg = fetch_service_links()
except Exception:
supported, links_cfg = {}, {}
_cell_dialog({
'feed_name': feed_name,
'date_col': date_col,
'raw_status': raw_status,
'days_back': _days_back(date_col) if date_col else None,
'updated_at': day.get('updated_at') or day.get('updated'),
'complete_status': day.get('complete_status'),
'swf_run_url': _swf_run_url(day) if date_col else None,
'links': get_links(feed_id, supported, links_cfg),
})
def _days_back(date_str):
"""Return integer days between today and an ISO date, or None on error."""
try:
target = date_cls.fromisoformat(date_str)
except ValueError:
return None
return (date_cls.today() - target).days
_render_nav()
top_left, top_right = st.columns([1, 4], vertical_alignment='bottom')
with top_left:
pick_date = st.date_input(
'Pick date to go back to', value=date_cls.today())
with top_right:
if st.button('Refresh', width='content'):
fetch_section.clear()
fetch_service_links.clear()
st.rerun()
date_arg = None
if isinstance(pick_date, date_cls) and pick_date != date_cls.today():
date_arg = pick_date.isoformat()
for section in SECTIONS:
st.divider()
st.subheader(section.title, anchor=section.id)
render_section(section, date_arg)