"""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)