import os import re from datetime import datetime import pandas as pd import plotly.express as px import plotly.graph_objects as go from dash import MATCH, Dash, Input, Output, State, dash_table, dcc, html from dash.exceptions import PreventUpdate # ── Constants ──────────────────────────────────────────────────────────────── ANNOTATION_FILE = "annotated_product_replacements.csv" ANNOTATION_COLS = [ "REJECTED_ITEM_ID", "APPROVED_ITEM_ID", "is_direct_replacement", "reasoning", "timestamp", ] # ── Load & clean data ──────────────────────────────────────────────────────── df = pd.read_csv("approved_after_with_comments_for_analysis.csv") STATUS_LABELS = { 0: "Draft", 1: "In Review", 2: "Selected", 3: "Quoting", 4: "Re-submit", 5: "Rejected", 7: "Approved", 8: "Ordered", 9: "Payment Due", 10: "In Production", 11: "In Transit", 12: "Installed", 13: "Delivered", 14: "Closed", 15: "Client Review", 16: "Hidden", 17: "Invoiced", 18: "Partial Payment", 19: "Paid", } STATUS_COLORS = { "Draft": "#BDC3C7", "In Review": "#3498DB", "Selected": "#1ABC9C", "Quoting": "#F39C12", "Re-submit": "#E67E22", "Rejected": "#E74C3C", "Approved": "#2ECC71", "Ordered": "#27AE60", "Payment Due": "#F1C40F", "In Production": "#8E44AD", "In Transit": "#2980B9", "Installed": "#16A085", "Delivered": "#4C9BE8", "Closed": "#7F8C8D", "Client Review": "#9B59B6", "Hidden": "#95A5A6", "Invoiced": "#D35400", "Partial Payment": "#E74C3C", "Paid": "#2ECC71", } def parse_timedelta(s): """Convert '17 days 18:43:45.241499' → total hours (float).""" if not isinstance(s, str): return None m = re.match(r"(-?\d+) days \+?(-?\d+):(\d+):(\d+)", s) if not m: return None days, hours, minutes, seconds = int(m[1]), int(m[2]), int(m[3]), float(m[4]) total = days * 24 + hours + minutes / 60 + seconds / 3600 return total df["approval_hours"] = df["time_to_approval"].apply(parse_timedelta) df["approval_days"] = df["approval_hours"].apply( lambda h: round(h / 24, 1) if h is not None else None ) df["added_days"] = ( df["time_to_approved_added"] .apply(parse_timedelta) .apply(lambda h: round(h / 24, 1) if h is not None else None) ) df["status_label"] = ( df["APPROVED_ITEM_STATUS"] .map(STATUS_LABELS) .fillna(df["APPROVED_ITEM_STATUS"].astype(str)) ) df["rejected_top_cat"] = df["REJECTED_ITEM_CATEGORY"].str.split(":").str[0] df["approved_top_cat"] = df["APPROVED_ITEM_CATEGORY"].str.split(":").str[0] unique_rejected = df.drop_duplicates("REJECTED_ITEM_ID") # ── Summary numbers ────────────────────────────────────────────────────────── n_rejections = df["REJECTED_ITEM_ID"].nunique() n_approved_alts = df["APPROVED_ITEM_ID"].nunique() n_resolved = unique_rejected["RESOLVED"].astype(str).str.lower().eq("true").sum() n_rooms = df["SUBSECTION_NAME"].nunique() # ── Annotation helpers ─────────────────────────────────────────────────────── def load_annotations(): """Load annotations from CSV, return dict keyed by 'rejected_id|approved_id'.""" if not os.path.exists(ANNOTATION_FILE): return {} try: ann_df = pd.read_csv(ANNOTATION_FILE, dtype=str).fillna("") return { f"{row['REJECTED_ITEM_ID']}|{row['APPROVED_ITEM_ID']}": { "is_direct": row["is_direct_replacement"], "reasoning": row.get("reasoning", ""), "timestamp": row.get("timestamp", ""), } for _, row in ann_df.iterrows() } except Exception: return {} def write_annotation(rejected_id, approved_id, is_direct, reasoning): """Upsert annotation row in CSV, return updated annotations dict.""" timestamp = datetime.now().isoformat(timespec="seconds") new_row = { "REJECTED_ITEM_ID": str(rejected_id), "APPROVED_ITEM_ID": str(approved_id), "is_direct_replacement": is_direct, "reasoning": reasoning or "", "timestamp": timestamp, } if os.path.exists(ANNOTATION_FILE): ann_df = pd.read_csv(ANNOTATION_FILE, dtype=str) mask = (ann_df["REJECTED_ITEM_ID"] == str(rejected_id)) & ( ann_df["APPROVED_ITEM_ID"] == str(approved_id) ) if mask.any(): for col, val in new_row.items(): ann_df.loc[mask, col] = val else: ann_df = pd.concat([ann_df, pd.DataFrame([new_row])], ignore_index=True) else: ann_df = pd.DataFrame([new_row], columns=ANNOTATION_COLS) ann_df.to_csv(ANNOTATION_FILE, index=False) return load_annotations() def annotation_status_by_schedule(annotations, filter_category=True): """Return (fully_done, partial) sets of SCHEDULE_ITEM_IDs.""" fully_done, partial = set(), set() for sid in df["SCHEDULE_ITEM_ID"].unique(): rows = df[df["SCHEDULE_ITEM_ID"] == sid] if filter_category: rej_cat = rows.iloc[0]["REJECTED_ITEM_CATEGORY"] rows = rows[rows["APPROVED_ITEM_CATEGORY"] == rej_cat] keys = { f"{str(r['REJECTED_ITEM_ID'])}|{str(r['APPROVED_ITEM_ID'])}" for _, r in rows.iterrows() } if not keys: continue n_done = len(keys & annotations.keys()) if n_done == len(keys): fully_done.add(str(sid)) elif n_done > 0: partial.add(str(sid)) return fully_done, partial # ── Shared helpers ─────────────────────────────────────────────────────────── CARD_STYLE = { "background": "#fff", "borderRadius": 8, "padding": "16px 20px", "boxShadow": "0 1px 4px rgba(0,0,0,.08)", "marginBottom": 20, } PAGE_STYLE = { "fontFamily": "Inter, system-ui, sans-serif", "background": "#f8f9fa", "minHeight": "100vh", "padding": "24px 32px", } def stat_card(label, value, color="#4C9BE8"): return html.Div( [ html.Div( str(value), style={"fontSize": 36, "fontWeight": 700, "color": color} ), html.Div(label, style={"fontSize": 13, "color": "#888", "marginTop": 2}), ], style={ "background": "#fff", "borderRadius": 8, "padding": "16px 24px", "boxShadow": "0 1px 4px rgba(0,0,0,.08)", "minWidth": 140, }, ) def nav_bar(active): link_style = { "textDecoration": "none", "fontSize": 14, "fontWeight": 500, "padding": "6px 14px", "borderRadius": 6, } active_style = {**link_style, "background": "#4C9BE8", "color": "#fff"} inactive_style = {**link_style, "color": "#555"} return html.Div( [ html.Span( "Replacement Products", style={ "fontWeight": 700, "fontSize": 16, "color": "#1a1a2e", "marginRight": 24, }, ), dcc.Link( "Overview", href="/", style=active_style if active == "overview" else inactive_style, ), dcc.Link( "Item Detail", href="/item", style=active_style if active == "item" else inactive_style, ), dcc.Link( "Analysis", href="/analysis", style=active_style if active == "analysis" else inactive_style, ), ], style={ "display": "flex", "alignItems": "center", "gap": 8, "background": "#fff", "padding": "12px 24px", "borderRadius": 8, "boxShadow": "0 1px 4px rgba(0,0,0,.08)", "marginBottom": 24, }, ) def label_row(label, value): if not value or (isinstance(value, float) and pd.isna(value)): return None return html.Div( [ html.Span( label, style={ "fontSize": 11, "fontWeight": 600, "color": "#999", "textTransform": "uppercase", "letterSpacing": "0.05em", "display": "block", "marginBottom": 2, }, ), html.Span(str(value), style={"fontSize": 13, "color": "#222"}), ], style={"marginBottom": 12}, ) def status_badge(label): color = STATUS_COLORS.get(label, "#95A5A6") return html.Span( label, style={ "background": color, "color": "#fff", "fontSize": 11, "fontWeight": 600, "padding": "3px 10px", "borderRadius": 20, "display": "inline-block", }, ) # ── Overview charts ────────────────────────────────────────────────────────── reason_counts = ( df.groupby("reason")["REJECTED_ITEM_ID"] .nunique() .sort_values(ascending=True) .tail(20) ) fig_reasons = px.bar( x=reason_counts.values, y=reason_counts.index, orientation="h", labels={"x": "# Rejected Items", "y": ""}, title="Top Rejection Reasons", color=reason_counts.values, color_continuous_scale="Blues", ) fig_reasons.update_layout( coloraxis_showscale=False, margin=dict(l=10, r=20, t=40, b=10), height=500, paper_bgcolor="#f8f9fa", plot_bgcolor="#f8f9fa", ) status_counts = df["status_label"].value_counts() fig_status = px.pie( values=status_counts.values, names=status_counts.index, title="Approved Alternative Status", hole=0.45, color_discrete_sequence=px.colors.qualitative.Pastel, ) fig_status.update_layout( margin=dict(l=10, r=10, t=40, b=10), height=360, paper_bgcolor="#f8f9fa" ) fig_time = px.histogram( df["approval_days"].dropna(), nbins=40, title="Time to Approval (days)", labels={"value": "Days"}, color_discrete_sequence=["#4C9BE8"], ) fig_time.update_layout( showlegend=False, margin=dict(l=10, r=10, t=40, b=10), height=320, paper_bgcolor="#f8f9fa", plot_bgcolor="#f8f9fa", xaxis_title="Days to Approval", yaxis_title="Count", ) sankey_df = ( df.groupby(["rejected_top_cat", "approved_top_cat"]) .size() .reset_index(name="count") .query("count >= 2") ) all_nodes = list( pd.unique( sankey_df["rejected_top_cat"].tolist() + sankey_df["approved_top_cat"].tolist() ) ) node_idx = {n: i for i, n in enumerate(all_nodes)} n_rej_nodes = len(sankey_df["rejected_top_cat"].unique()) fig_sankey = go.Figure( go.Sankey( node=dict( label=all_nodes, pad=12, thickness=18, color=["#4C9BE8"] * n_rej_nodes + ["#F4A261"] * (len(all_nodes) - n_rej_nodes), ), link=dict( source=[node_idx[r] for r in sankey_df["rejected_top_cat"]], target=[node_idx[a] for a in sankey_df["approved_top_cat"]], value=sankey_df["count"].tolist(), color="rgba(76,155,232,0.25)", ), ) ) fig_sankey.update_layout( title="Rejected → Approved Category Flow", height=420, margin=dict(l=10, r=10, t=40, b=10), paper_bgcolor="#f8f9fa", ) room_counts = ( df.groupby("SUBSECTION_NAME")["REJECTED_ITEM_ID"] .nunique() .sort_values(ascending=False) .head(15) ) fig_rooms = px.bar( x=room_counts.index, y=room_counts.values, title="Rejections by Room / Subsection", labels={"x": "", "y": "# Rejected Items"}, color=room_counts.values, color_continuous_scale="Teal", ) fig_rooms.update_layout( coloraxis_showscale=False, margin=dict(l=10, r=10, t=40, b=30), height=320, paper_bgcolor="#f8f9fa", plot_bgcolor="#f8f9fa", xaxis_tickangle=-35, ) TABLE_COLS = [ "REJECTED_ITEM_NAME", "REJECTED_ITEM_CATEGORY", "COMMENT", "reason", "APPROVED_ITEM_NAME", "APPROVED_ITEM_CATEGORY", "status_label", "APPROVED_ITEM_COMMENT", "approval_days", "SUBSECTION_NAME", "confidence", ] table_df = df[TABLE_COLS].rename( columns={ "REJECTED_ITEM_NAME": "Rejected Item", "REJECTED_ITEM_CATEGORY": "Rejected Category", "COMMENT": "Rejection Comment", "reason": "Reason", "APPROVED_ITEM_NAME": "Approved Alternative", "APPROVED_ITEM_CATEGORY": "Approved Category", "status_label": "Status", "APPROVED_ITEM_COMMENT": "Approval Comment", "approval_days": "Days to Approval", "SUBSECTION_NAME": "Room", "confidence": "Confidence", } ) # ── Overview layout ────────────────────────────────────────────────────────── def overview_layout(): return html.Div( style=PAGE_STYLE, children=[ nav_bar("overview"), html.Div( [ stat_card("Rejected Items", n_rejections, "#4C9BE8"), stat_card("Approved Alternatives", n_approved_alts, "#2ECC71"), stat_card("Rejections Resolved", n_resolved, "#F4A261"), stat_card("Rooms / Sections", n_rooms, "#9B59B6"), ], style={ "display": "flex", "gap": 16, "marginBottom": 24, "flexWrap": "wrap", }, ), html.Div( [ html.Div( dcc.Graph(figure=fig_reasons), style={**CARD_STYLE, "flex": "2"} ), html.Div( dcc.Graph(figure=fig_status), style={**CARD_STYLE, "flex": "1"} ), ], style={"display": "flex", "gap": 16}, ), html.Div(dcc.Graph(figure=fig_sankey), style=CARD_STYLE), html.Div( [ html.Div( dcc.Graph(figure=fig_time), style={**CARD_STYLE, "flex": "1"} ), html.Div( dcc.Graph(figure=fig_rooms), style={**CARD_STYLE, "flex": "1"} ), ], style={"display": "flex", "gap": 16}, ), html.Div( [ html.H3( "Browse Replacement Products", style={"marginTop": 0, "fontSize": 16, "fontWeight": 600}, ), html.Div( [ dcc.Input( id="search-input", type="text", placeholder="Search rejected or approved item...", debounce=True, style={ "width": "340px", "padding": "8px 12px", "borderRadius": 6, "border": "1px solid #ddd", "fontSize": 13, }, ), dcc.Dropdown( id="status-filter", options=[ {"label": s, "value": s} for s in sorted(table_df["Status"].unique()) ], placeholder="Filter by status…", multi=True, style={"width": 260, "fontSize": 13}, ), dcc.Dropdown( id="room-filter", options=[ {"label": r, "value": r} for r in sorted(table_df["Room"].dropna().unique()) ], placeholder="Filter by room…", multi=True, style={"width": 240, "fontSize": 13}, ), ], style={ "display": "flex", "alignItems": "center", "marginBottom": 12, "flexWrap": "wrap", "gap": 8, }, ), dash_table.DataTable( id="main-table", columns=[{"name": c, "id": c} for c in table_df.columns], data=table_df.to_dict("records"), page_size=20, sort_action="native", style_table={"overflowX": "auto"}, style_header={ "backgroundColor": "#f0f4f8", "fontWeight": 600, "fontSize": 12, "border": "none", "padding": "10px 12px", }, style_cell={ "fontSize": 12, "padding": "8px 12px", "border": "none", "borderBottom": "1px solid #f0f0f0", "maxWidth": 260, "overflow": "hidden", "textOverflow": "ellipsis", "whiteSpace": "nowrap", }, style_data_conditional=[ {"if": {"row_index": "odd"}, "backgroundColor": "#fafbfc"}, { "if": {"filter_query": '{Status} = "Approved"'}, "backgroundColor": "#f0fdf4", }, ], tooltip_data=[ { col: { "value": str(row[col]) if row[col] is not None else "", "type": "markdown", } for col in table_df.columns } for row in table_df.to_dict("records") ], tooltip_duration=None, ), ], style=CARD_STYLE, ), ], ) # ── Item detail helpers ────────────────────────────────────────────────────── item_options_df = ( df[["SCHEDULE_ITEM_ID", "REJECTED_ITEM_NAME", "SUBSECTION_NAME"]] .drop_duplicates("SCHEDULE_ITEM_ID") .sort_values("REJECTED_ITEM_NAME") ) def build_item_dropdown_options(annotations, filter_category=True): fully_done, partial = annotation_status_by_schedule(annotations, filter_category) options = [] for row in item_options_df.itertuples(): sid = str(row.SCHEDULE_ITEM_ID) name = row.REJECTED_ITEM_NAME or "Unknown" room = row.SUBSECTION_NAME or "" s_rows = df[df["SCHEDULE_ITEM_ID"] == row.SCHEDULE_ITEM_ID] if filter_category: rej_cat = s_rows.iloc[0]["REJECTED_ITEM_CATEGORY"] s_rows = s_rows[s_rows["APPROVED_ITEM_CATEGORY"] == rej_cat] n_total = len(s_rows) keys = { f"{str(r['REJECTED_ITEM_ID'])}|{str(r['APPROVED_ITEM_ID'])}" for _, r in s_rows.iterrows() } n_annotated = len(keys & annotations.keys()) if sid in fully_done: prefix = "✅ " elif sid in partial: prefix = "❗ " else: prefix = "" options.append( { "label": f"{prefix}{name} — {room} ({n_annotated}/{n_total})", "value": sid, } ) return options def annotation_controls(key, existing): """Annotation UI block for one approved item card.""" pre_value = existing.get("is_direct") if existing else None pre_reasoning = existing.get("reasoning", "") if existing else "" pre_ts = existing.get("timestamp", "") if existing else "" already_badge = None if existing: if pre_value == "true": label, badge_color = "✓ Direct Replacement", "#2ECC71" elif pre_value == "false": label, badge_color = "✗ Not Direct", "#E74C3C" else: label, badge_color = "? Needs SME", "#F39C12" already_badge = html.Div( [ html.Span( label, style={ "background": badge_color, "color": "#fff", "fontSize": 11, "fontWeight": 600, "padding": "3px 10px", "borderRadius": 20, "display": "inline-block", "marginRight": 8, }, ), html.Span( f"annotated {pre_ts[:10]}" if pre_ts else "", style={"fontSize": 11, "color": "#aaa"}, ), ], style={"marginBottom": 8}, ) return html.Div( [ html.Div( [ html.Span( "Annotation", style={ "fontSize": 11, "fontWeight": 700, "color": "#999", "textTransform": "uppercase", "letterSpacing": "0.05em", }, ), html.Hr(style={"margin": "6px 0 10px", "borderColor": "#f0f0f0"}), ], ), already_badge, dcc.RadioItems( id={"type": "is-direct-radio", "index": key}, options=[ {"label": " ✓ Direct Replacement", "value": "true"}, {"label": " ✗ Not Direct", "value": "false"}, {"label": " ? Needs SME", "value": "sme"}, ], value=pre_value, inline=True, inputStyle={"marginRight": 4}, labelStyle={"marginRight": 20, "fontSize": 13, "cursor": "pointer"}, ), dcc.Textarea( id={"type": "reasoning-input", "index": key}, value=pre_reasoning, placeholder="Reasoning (optional)…", style={ "width": "100%", "height": 56, "fontSize": 12, "borderRadius": 4, "border": "1px solid #ddd", "padding": "6px 8px", "resize": "vertical", "marginTop": 8, "boxSizing": "border-box", }, ), html.Div( [ html.Button( "Confirm", id={"type": "confirm-btn", "index": key}, n_clicks=0, style={ "background": "#4C9BE8", "color": "#fff", "border": "none", "borderRadius": 4, "padding": "6px 18px", "cursor": "pointer", "fontSize": 12, "marginTop": 8, }, ), html.Span( id={"type": "confirm-feedback", "index": key}, style={"fontSize": 11, "marginLeft": 10, "color": "#2ECC71"}, ), ], style={"display": "flex", "alignItems": "center"}, ), ], style={"borderTop": "1px solid #f0f0f0", "marginTop": 12, "paddingTop": 10}, ) def approved_card(row, annotations): key = ( f"{str(row.get('REJECTED_ITEM_ID', ''))}|{str(row.get('APPROVED_ITEM_ID', ''))}" ) existing = annotations.get(key) url = row.get("APPROVED_ITEM_URL", "") name = row.get("APPROVED_ITEM_NAME", "Unknown") name_el = ( html.A( name, href=url, target="_blank", style={ "fontSize": 14, "fontWeight": 600, "color": "#4C9BE8", "textDecoration": "none", }, ) if url and isinstance(url, str) and url.startswith("http") else html.Span( name, style={"fontSize": 14, "fontWeight": 600, "color": "#1a1a2e"} ) ) status = row.get("status_label", "") fields = [ f for f in [ label_row("Item ID", row.get("APPROVED_ITEM_ID")), label_row("Category", row.get("APPROVED_ITEM_CATEGORY")), label_row( "Added", str(row.get("APPROVED_ITEM_ADDED_AT", ""))[:10] if row.get("APPROVED_ITEM_ADDED_AT") else None, ), label_row( "Approved", str(row.get("APPROVED_ITEM_APPROVED_AT", ""))[:10] if row.get("APPROVED_ITEM_APPROVED_AT") and str(row.get("APPROVED_ITEM_APPROVED_AT")) != "nan" else None, ), label_row( "Days (rejection → added)", row.get("added_days") if str(row.get("added_days", "")) != "nan" else None, ), label_row( "Days (rejection → approved)", row.get("approval_days") if str(row.get("approval_days", "")) != "nan" else None, ), ] if f is not None ] comment = row.get("APPROVED_ITEM_COMMENT", "") if comment and isinstance(comment, str) and comment.strip(): fields.append( html.Div( [ html.Span( "Client Comment", style={ "fontSize": 11, "fontWeight": 600, "color": "#999", "textTransform": "uppercase", "letterSpacing": "0.05em", "display": "block", "marginBottom": 4, }, ), html.P( comment, style={ "fontSize": 13, "color": "#555", "margin": 0, "lineHeight": "1.5", "fontStyle": "italic", }, ), ], style={"marginBottom": 12}, ) ) annotated_indicator = "" if existing: annotated_indicator = "🟢 " return html.Div( [ html.Div( [ html.Span( annotated_indicator + name if annotated_indicator else None, style={"display": "none"}, ) if False else None, html.Div( [ name_el, html.Div(status_badge(status), style={"marginTop": 6}), ], style={"marginBottom": 10}, ), *fields, annotation_controls(key, existing), ] ), ], style={ "background": "#f8f9fa" if not existing else "#f0fdf4", "borderRadius": 8, "padding": "14px 16px", "marginBottom": 12, "borderLeft": f"3px solid {STATUS_COLORS.get(status, '#ddd')}", "outline": "2px solid #2ECC71" if existing else "none", }, ) # ── Item detail layout ─────────────────────────────────────────────────────── def item_detail_layout(): annotations = load_annotations() n_annotated = len(annotations) n_total = df["APPROVED_ITEM_ID"].nunique() return html.Div( style=PAGE_STYLE, children=[ dcc.Store(id="annotations-store", data=annotations), dcc.Interval(id="annotation-poll", interval=1500, n_intervals=0), # Nav + counter html.Div( [ nav_bar("item"), html.Div( [ html.Span( id="annotation-counter", children=f"{n_annotated} / {n_total} annotated", style={ "fontSize": 13, "fontWeight": 600, "color": "#fff", "background": "#2ECC71" if n_annotated > 0 else "#aaa", "padding": "6px 14px", "borderRadius": 20, }, ), ], style={ "position": "fixed", "top": 24, "right": 32, "zIndex": 1000, }, ), ] ), # Item selector + category filter toggle html.Div( [ dcc.Dropdown( id="item-select", options=build_item_dropdown_options(annotations), placeholder="Select a rejected item…", clearable=True, style={"fontSize": 14, "flex": 1}, ), dcc.Checklist( id="category-filter-toggle", options=[{"label": " Match category only", "value": "filter"}], value=["filter"], inputStyle={"marginRight": 5}, labelStyle={ "fontSize": 13, "color": "#555", "cursor": "pointer", "whiteSpace": "nowrap", }, ), ], style={ **CARD_STYLE, "display": "flex", "alignItems": "center", "gap": 16, }, ), html.Div(id="item-detail-content"), ], ) # ── App setup ──────────────────────────────────────────────────────────────── app = Dash(__name__, suppress_callback_exceptions=True) app.title = "Replacement Products Viewer" app.layout = html.Div( [ dcc.Location(id="url", refresh=False), html.Div(id="page-content"), ] ) # ── Routing ────────────────────────────────────────────────────────────────── def analysis_layout(): return html.Div( style=PAGE_STYLE, children=[ nav_bar("analysis"), html.Div( [ dcc.Checklist( id="analysis-cat-filter", options=[{"label": " Match category only", "value": "filter"}], value=["filter"], inputStyle={"marginRight": 5}, labelStyle={ "fontSize": 13, "color": "#555", "cursor": "pointer", "whiteSpace": "nowrap", }, ), ], style={ **CARD_STYLE, "display": "flex", "alignItems": "center", "gap": 12, }, ), html.Div(id="analysis-content"), ], ) @app.callback( Output("analysis-content", "children"), Input("analysis-cat-filter", "value"), ) def render_analysis_content(cat_filter): annotations = load_annotations() if not annotations: return html.Div( "No annotations yet. Annotate some items on the Item Detail page first.", style={ "color": "#aaa", "fontSize": 14, "textAlign": "center", "padding": "60px 0", }, ) filter_on = cat_filter and "filter" in cat_filter # ── Build annotated dataframe ────────────────────────────────────────── ann_rows = [] for key, val in annotations.items(): rejected_id, approved_id = key.split("|") ann_rows.append( { "REJECTED_ITEM_ID": rejected_id, "APPROVED_ITEM_ID": approved_id, "is_direct": val["is_direct"], "reasoning": val["reasoning"], "timestamp": val["timestamp"], } ) ann_df = pd.DataFrame(ann_rows) ann_df["REJECTED_ITEM_ID"] = ann_df["REJECTED_ITEM_ID"].astype(int) ann_df["APPROVED_ITEM_ID"] = ann_df["APPROVED_ITEM_ID"].astype(int) merged = ann_df.merge( df[ [ "REJECTED_ITEM_ID", "APPROVED_ITEM_ID", "REJECTED_ITEM_CATEGORY", "APPROVED_ITEM_CATEGORY", "reason", "SUBSECTION_NAME", "approval_days", "added_days", "status_label", "REJECTED_ITEM_NAME", "APPROVED_ITEM_NAME", "APPROVED_ITEM_COMMENT", ] ], on=["REJECTED_ITEM_ID", "APPROVED_ITEM_ID"], how="left", ) merged["rejected_top_cat"] = merged["REJECTED_ITEM_CATEGORY"].str.split(":").str[0] if filter_on: merged = merged[ merged["APPROVED_ITEM_CATEGORY"] == merged["REJECTED_ITEM_CATEGORY"] ] LABEL_MAP = { "true": "Direct Replacement", "false": "Not Direct", "sme": "Needs SME", } COLOR_MAP = { "Direct Replacement": "#2ECC71", "Not Direct": "#E74C3C", "Needs SME": "#F39C12", } merged["label"] = merged["is_direct"].map(LABEL_MAP).fillna("Unknown") # Unannotated annotated_keys = set(annotations.keys()) unannotated_rows = [] for sid in df["SCHEDULE_ITEM_ID"].unique(): s_rows = df[df["SCHEDULE_ITEM_ID"] == sid] rej_cat = s_rows.iloc[0]["REJECTED_ITEM_CATEGORY"] candidates = ( s_rows[s_rows["APPROVED_ITEM_CATEGORY"] == rej_cat] if filter_on else s_rows ) for _, r in candidates.iterrows(): key = f"{str(int(r['REJECTED_ITEM_ID']))}|{str(int(r['APPROVED_ITEM_ID']))}" if key not in annotated_keys: unannotated_rows.append( { "Rejected Item ID": int(r["REJECTED_ITEM_ID"]), "Rejected Item": r["REJECTED_ITEM_NAME"], "Approved Item ID": int(r["APPROVED_ITEM_ID"]), "Approved Item": r["APPROVED_ITEM_NAME"], "Category": rej_cat, "Room": r["SUBSECTION_NAME"], "Status": r["status_label"], "Days to Approval": r["approval_days"], "Rejection Comment": r["COMMENT"], "Reason": r["reason"], } ) unannotated_df = pd.DataFrame(unannotated_rows) if filter_on: n_total = sum( len( df[ (df["SCHEDULE_ITEM_ID"] == sid) & ( df["APPROVED_ITEM_CATEGORY"] == df[df["SCHEDULE_ITEM_ID"] == sid].iloc[0][ "REJECTED_ITEM_CATEGORY" ] ) ] ) for sid in df["SCHEDULE_ITEM_ID"].unique() ) else: n_total = df["APPROVED_ITEM_ID"].nunique() n_done = len(merged) n_direct = (merged["is_direct"] == "true").sum() n_not_direct = (merged["is_direct"] == "false").sum() n_sme = (merged["is_direct"] == "sme").sum() if merged.empty: return html.Div( "No annotated data for the selected categories.", style={ "color": "#aaa", "fontSize": 14, "textAlign": "center", "padding": "40px 0", }, ) # ── Charts ───────────────────────────────────────────────────────────── fig_donut = px.pie( values=[n_direct, n_not_direct, n_sme, max(n_total - n_done, 0)], names=["Direct Replacement", "Not Direct", "Needs SME", "Unannotated"], hole=0.55, title="Annotation Breakdown", color_discrete_sequence=["#2ECC71", "#E74C3C", "#F39C12", "#E0E0E0"], ) fig_donut.update_layout( margin=dict(l=10, r=10, t=40, b=10), height=340, paper_bgcolor="#f8f9fa" ) cat_summary = ( merged.groupby(["rejected_top_cat", "label"]).size().reset_index(name="count") ) fig_cat = px.bar( cat_summary, x="rejected_top_cat", y="count", color="label", title="Annotation Result by Category", labels={"rejected_top_cat": "", "count": "# Items", "label": ""}, color_discrete_map=COLOR_MAP, barmode="stack", ) fig_cat.update_layout( margin=dict(l=10, r=10, t=40, b=60), height=360, paper_bgcolor="#f8f9fa", plot_bgcolor="#f8f9fa", xaxis_tickangle=-30, legend=dict(orientation="h", y=1.1), ) reason_summary = ( merged.groupby(["reason", "label"]).size().reset_index(name="count") ) top_reasons = ( reason_summary.groupby("reason")["count"] .sum() .sort_values(ascending=False) .head(15) .index ) reason_summary = reason_summary[reason_summary["reason"].isin(top_reasons)] fig_reason = px.bar( reason_summary, x="count", y="reason", color="label", orientation="h", title="Annotation Result by Rejection Reason (top 15)", labels={"reason": "", "count": "# Items", "label": ""}, color_discrete_map=COLOR_MAP, barmode="stack", ) fig_reason.update_layout( margin=dict(l=10, r=20, t=40, b=10), height=480, paper_bgcolor="#f8f9fa", plot_bgcolor="#f8f9fa", legend=dict(orientation="h", y=1.05), ) days_data = merged[merged["approval_days"].notna() & (merged["approval_days"] >= 0)] fig_days = px.box( days_data, x="label", y="approval_days", color="label", title="Days to Approval by Annotation Type", labels={"label": "", "approval_days": "Days"}, color_discrete_map=COLOR_MAP, points="outliers", ) fig_days.update_layout( showlegend=False, margin=dict(l=10, r=10, t=40, b=10), height=320, paper_bgcolor="#f8f9fa", plot_bgcolor="#f8f9fa", ) direct_status = merged[merged["is_direct"] == "true"]["status_label"].value_counts() fig_direct_status = px.bar( x=direct_status.index, y=direct_status.values, title="Status of Direct Replacements", labels={"x": "", "y": "# Items"}, color=direct_status.index, color_discrete_map=STATUS_COLORS, ) fig_direct_status.update_layout( showlegend=False, margin=dict(l=10, r=10, t=40, b=40), height=320, paper_bgcolor="#f8f9fa", plot_bgcolor="#f8f9fa", xaxis_tickangle=-20, ) table_cols = [ "REJECTED_ITEM_NAME", "APPROVED_ITEM_NAME", "REJECTED_ITEM_CATEGORY", "label", "reasoning", "status_label", "approval_days", "SUBSECTION_NAME", "timestamp", ] table_data = ( merged[table_cols] .rename( columns={ "REJECTED_ITEM_NAME": "Rejected Item", "APPROVED_ITEM_NAME": "Approved Item", "REJECTED_ITEM_CATEGORY": "Category", "label": "Annotation", "reasoning": "Reasoning", "status_label": "Status", "approval_days": "Days to Approval", "SUBSECTION_NAME": "Room", "timestamp": "Annotated At", } ) .sort_values("Annotation") ) def make_table(data, cond_styles=None): return dash_table.DataTable( columns=[{"name": c, "id": c} for c in data.columns], data=data.to_dict("records"), page_size=20, sort_action="native", filter_action="native", style_table={"overflowX": "auto"}, style_header={ "backgroundColor": "#f0f4f8", "fontWeight": 600, "fontSize": 12, "border": "none", "padding": "10px 12px", }, style_cell={ "fontSize": 12, "padding": "8px 12px", "border": "none", "borderBottom": "1px solid #f0f0f0", "maxWidth": 300, "overflow": "hidden", "textOverflow": "ellipsis", "whiteSpace": "nowrap", }, style_data_conditional=[ {"if": {"row_index": "odd"}, "backgroundColor": "#fafbfc"}, *(cond_styles or []), ], tooltip_data=[ { col: { "value": str(row[col]) if row[col] is not None else "", "type": "markdown", } for col in data.columns } for row in data.to_dict("records") ], tooltip_duration=None, ) return html.Div( [ html.Div( [ stat_card("Annotated", f"{n_done} / {n_total}", "#4C9BE8"), stat_card("Direct Replacements", n_direct, "#2ECC71"), stat_card("Not Direct", n_not_direct, "#E74C3C"), stat_card("Needs SME", n_sme, "#F39C12"), ], style={ "display": "flex", "gap": 16, "marginBottom": 24, "flexWrap": "wrap", }, ), html.Div( [ html.Div( dcc.Graph(figure=fig_donut), style={**CARD_STYLE, "flex": "1"} ), html.Div( dcc.Graph(figure=fig_days), style={**CARD_STYLE, "flex": "1"} ), ], style={"display": "flex", "gap": 16}, ), html.Div(dcc.Graph(figure=fig_cat), style=CARD_STYLE), html.Div( [ html.Div( dcc.Graph(figure=fig_reason), style={**CARD_STYLE, "flex": "2"} ), html.Div( dcc.Graph(figure=fig_direct_status), style={**CARD_STYLE, "flex": "1"}, ), ], style={"display": "flex", "gap": 16}, ), html.Div( [ html.H3( "Annotation Log", style={"marginTop": 0, "fontSize": 16, "fontWeight": 600}, ), make_table( table_data, [ { "if": { "filter_query": '{Annotation} = "Direct Replacement"' }, "backgroundColor": "#f0fdf4", }, { "if": {"filter_query": '{Annotation} = "Not Direct"'}, "backgroundColor": "#fef2f2", }, { "if": {"filter_query": '{Annotation} = "Needs SME"'}, "backgroundColor": "#fffbeb", }, ], ), ], style=CARD_STYLE, ), html.Div( [ html.H3( f"Unannotated Items ({len(unannotated_df)})", style={"marginTop": 0, "fontSize": 16, "fontWeight": 600}, ), make_table(unannotated_df) if not unannotated_df.empty else html.Div( "All items annotated.", style={"color": "#aaa", "fontSize": 13} ), ], style=CARD_STYLE, ), ] ) @app.callback(Output("page-content", "children"), Input("url", "pathname")) def display_page(pathname): if pathname == "/item": return item_detail_layout() if pathname == "/analysis": return analysis_layout() return overview_layout() # ── Overview table filter ──────────────────────────────────────────────────── @app.callback( Output("main-table", "data"), Input("search-input", "value"), Input("status-filter", "value"), Input("room-filter", "value"), ) def filter_table(search, statuses, rooms): filtered = table_df.copy() if search: mask = ( filtered["Rejected Item"].str.contains(search, case=False, na=False) | filtered["Approved Alternative"].str.contains( search, case=False, na=False ) | filtered["Reason"].str.contains(search, case=False, na=False) | filtered["Rejection Comment"].str.contains(search, case=False, na=False) ) filtered = filtered[mask] if statuses: filtered = filtered[filtered["Status"].isin(statuses)] if rooms: filtered = filtered[filtered["Room"].isin(rooms)] return filtered.to_dict("records") # ── Item detail render ─────────────────────────────────────────────────────── @app.callback( Output("item-detail-content", "children"), Input("item-select", "value"), Input("category-filter-toggle", "value"), State("annotations-store", "data"), ) def render_item_detail(schedule_item_id, cat_filter, annotations): annotations = annotations or {} if not schedule_item_id: return html.Div( "Select an item above to see the rejected product and its approved alternatives.", style={ "color": "#aaa", "fontSize": 14, "textAlign": "center", "padding": "60px 0", }, ) rows = df[df["SCHEDULE_ITEM_ID"].astype(str) == schedule_item_id] if rows.empty: return html.Div("No data found.", style={"color": "#aaa", "fontSize": 14}) first = rows.iloc[0] rej_cat = first.get("REJECTED_ITEM_CATEGORY", "") if cat_filter and "filter" in cat_filter: rows = rows[rows["APPROVED_ITEM_CATEGORY"] == rej_cat] # Left panel — rejected item rej_url = first.get("REJECTED_ITEM_URL", "") rej_name = first.get("REJECTED_ITEM_NAME", "Unknown") name_el = ( html.A( rej_name, href=rej_url, target="_blank", style={ "fontSize": 18, "fontWeight": 700, "color": "#E74C3C", "textDecoration": "none", }, ) if rej_url and isinstance(rej_url, str) and rej_url.startswith("http") else html.H2( rej_name, style={ "fontSize": 18, "fontWeight": 700, "color": "#1a1a2e", "margin": "0 0 12px 0", }, ) ) resolved = str(first.get("RESOLVED", "")).lower() == "true" resolved_badge = html.Span( "Resolved" if resolved else "Unresolved", style={ "background": "#2ECC71" if resolved else "#E74C3C", "color": "#fff", "fontSize": 11, "fontWeight": 600, "padding": "3px 10px", "borderRadius": 20, "display": "inline-block", "marginBottom": 12, }, ) rejected_fields = [ f for f in [ label_row("Item ID", first.get("REJECTED_ITEM_ID")), label_row("Brand", first.get("BRAND")), label_row("Category", first.get("REJECTED_ITEM_CATEGORY")), label_row("Rejection Category", first.get("category")), label_row("Description", first.get("PRODUCT_DESCRIPTION")), label_row("Room", first.get("SUBSECTION_NAME")), label_row("Subsection ID", first.get("SUBSECTION_ID")), label_row("Status", first.get("STATUS")), label_row("Rejected", str(first.get("REJECTED_AT", ""))[:10]), label_row("Rejection Comment", first.get("COMMENT")), label_row("Reason (AI)", first.get("reason")), label_row("Alternative criteria", first.get("alternative")), label_row("Confidence", first.get("confidence")), ] if f is not None ] left_panel = html.Div( [ html.Div( [ html.Span( "REJECTED", style={ "fontSize": 10, "fontWeight": 700, "color": "#E74C3C", "letterSpacing": "0.1em", "display": "block", "marginBottom": 6, }, ), name_el, resolved_badge, ], style={"marginBottom": 12}, ), *rejected_fields, ], style={ **CARD_STYLE, "borderTop": "3px solid #E74C3C", "flex": "1", "minWidth": 280, }, ) # Right panel — approved alternatives n_alts = len(rows) cards = [approved_card(r, annotations) for r in rows.to_dict("records")] right_panel = html.Div( [ html.Div( html.Span( f"{n_alts} Approved Alternative{'s' if n_alts != 1 else ''}", style={"fontSize": 16, "fontWeight": 600, "color": "#1a1a2e"}, ), style={"marginBottom": 14}, ), *cards, ], style={ **CARD_STYLE, "borderTop": "3px solid #2ECC71", "flex": "2", "minWidth": 320, }, ) return html.Div( [left_panel, right_panel], style={ "display": "flex", "gap": 16, "alignItems": "flex-start", "flexWrap": "wrap", }, ) # ── Annotation save callback (MATCH-only outputs) ──────────────────────────── @app.callback( Output({"type": "confirm-feedback", "index": MATCH}, "children"), Input({"type": "confirm-btn", "index": MATCH}, "n_clicks"), State({"type": "is-direct-radio", "index": MATCH}, "value"), State({"type": "reasoning-input", "index": MATCH}, "value"), prevent_initial_call=True, ) def save_annotation_cb(n_clicks, is_direct, reasoning): if not n_clicks: raise PreventUpdate from dash import ctx key = ctx.triggered_id["index"] rejected_id, approved_id = key.split("|") write_annotation(rejected_id, approved_id, is_direct, reasoning) ts = datetime.now().strftime("%H:%M:%S") return f"✓ Saved at {ts}" # ── Poll annotations from file into store ──────────────────────────────────── @app.callback( Output("annotations-store", "data"), Input("annotation-poll", "n_intervals"), ) def poll_annotations(n): return load_annotations() # ── Update counter and dropdown after annotation ───────────────────────────── @app.callback( Output("annotation-counter", "children"), Output("annotation-counter", "style"), Output("item-select", "options"), Input("annotations-store", "data"), Input("category-filter-toggle", "value"), ) def update_counter_and_dropdown(annotations, cat_filter): annotations = annotations or {} n_annotated = len(annotations) filter_on = cat_filter and "filter" in cat_filter if filter_on: # Count only same-category approved items n_total = sum( len( df[ (df["SCHEDULE_ITEM_ID"] == sid) & ( df["APPROVED_ITEM_CATEGORY"] == df[df["SCHEDULE_ITEM_ID"] == sid].iloc[0][ "REJECTED_ITEM_CATEGORY" ] ) ] ) for sid in df["SCHEDULE_ITEM_ID"].unique() ) else: n_total = df["APPROVED_ITEM_ID"].nunique() counter_style = { "fontSize": 13, "fontWeight": 600, "color": "#fff", "background": "#2ECC71" if n_annotated > 0 else "#aaa", "padding": "6px 14px", "borderRadius": 20, } return ( f"{n_annotated} / {n_total} annotated", counter_style, build_item_dropdown_options(annotations, filter_category=filter_on), ) if __name__ == "__main__": app.run(debug=True, port=8050)