""" app.py — Fendt PESTEL-EL Sentinel: Strategic War Room Dashboard =============================================================== Architecture rules (CLAUDE.md): - Imports only from core/. Zero business logic in callbacks. - Callbacks are pure functions. No global state mutation. - All Plotly colors use rgba() — never 8-char hex (#rrggbbaa). - All chart builders tested in _preflight() before Dash starts. - Astra DB access only via SignalDB. Never call astrapy directly. - Styling via CSS className. Inline style only for dynamic values. Sponsor requirements implemented: 1. Verifiable source hyperlinks on every signal row. 2. Bold 12M/24M/36M time-horizon rings; Urgency Matrix at Overview top. 3. Universal strategic LLM prompt — no company-specific copy. 4. Default radar filter ≥ 0.50 (HIGH+CRITICAL only). 5. dcc.Interval wired to DB; sidebar + canvas auto-refresh every 6h. """ from __future__ import annotations import atexit import hashlib import os import sys import textwrap import threading as _threading from collections import defaultdict from datetime import datetime, timezone from pathlib import Path from typing import Optional # ── Silence noisy loggers before Dash import ────────────────── import logging logging.getLogger("werkzeug").setLevel(logging.ERROR) logging.getLogger("dash").setLevel(logging.ERROR) import json import dash import dash_bootstrap_components as dbc import dash_cytoscape as cyto import diskcache from flask_caching import Cache as _FlaskCache # ── Optional PDF/Markdown rendering ─────────────────────────── try: import markdown as _md_lib import fpdf as _fpdf_lib _PDF_OK = True except ImportError: _PDF_OK = False import numpy as np import plotly.graph_objects as go from dash import Input, Output, State, callback_context, dcc, html, no_update from dash import DiskcacheManager # ── Load .env ───────────────────────────────────────────────── try: from dotenv import load_dotenv load_dotenv(override=False) except ImportError: _env = Path(__file__).parent / ".env" if _env.exists(): for line in _env.read_text().splitlines(): line = line.strip() if line and not line.startswith("#") and "=" in line: k, _, v = line.partition("=") os.environ.setdefault(k.strip(), v.strip()) sys.path.insert(0, str(Path(__file__).parent)) from core.database import PESTELDimension, Signal, SignalDB from core.scheduler import HEALTH, engine as _scheduler_engine from core.logger import get_logger from core.summary_engine import generate_brief_markdown from core.agents import run_agent_query from core.graph_engine import get_causal_chains, rebuild_graph_from_db, infer_hidden_relationships log = get_logger(__name__) # ───────────────────────────────────────────────────────────── # HuggingFace — API token + model config # ───────────────────────────────────────────────────────────── _HF_TOKEN = os.getenv("HUGGINGFACEHUB_API_TOKEN", "") _HF_OK = bool(_HF_TOKEN) _HF_REPO_ID = "meta-llama/Llama-3.1-8B-Instruct" # ───────────────────────────────────────────────────────────── # DB singleton — all access via SignalDB (CLAUDE.md rule) # ───────────────────────────────────────────────────────────── _db: Optional[SignalDB] = None def _get_db() -> SignalDB: global _db if _db is None: _db = SignalDB() return _db # _db_stats() replaced by _db_stats_cached() (Flask-Caching, see below) # ───────────────────────────────────────────────────────────── # Design Tokens # ───────────────────────────────────────────────────────────── _CAT_COLOUR = { "POLITICAL": "#64b5f6", "ECONOMIC": "#a5d6a7", "SOCIAL": "#ffcc80", "TECHNOLOGICAL": "#ce93d8", "ENVIRONMENTAL": "#80deea", "LEGAL": "#ef9a9a", } _DIM_COLOUR = _CAT_COLOUR # alias — same mapping used by chart + UI helpers _DIM_PILL_CODE = { "POLITICAL": "P", "ECONOMIC": "E", "SOCIAL": "S", "TECHNOLOGICAL": "T", "ENVIRONMENTAL": "En", "LEGAL": "L", } _SEV_COLOUR = {"critical": "#ff1744", "high": "#ffab00", "moderate": "#00e5ff", "low": "#607d8b"} def _hex_to_rgba(hex_colour: str, alpha: float = 0.30) -> str: """Convert '#rrggbb' → 'rgba(r,g,b,alpha)'. Plotly rejects 8-char hex.""" h = hex_colour.lstrip("#") r, g, b = int(h[0:2], 16), int(h[2:4], 16), int(h[4:6], 16) return f"rgba({r},{g},{b},{alpha})" def _sev(score: float) -> str: if score >= 0.75: return "critical" if score >= 0.50: return "high" if score >= 0.30: return "moderate" return "low" # Shared Plotly layout defaults _CHART_BASE = dict( template="plotly_dark", paper_bgcolor="rgba(0,0,0,0)", plot_bgcolor="rgba(0,0,0,0)", font_family="Inter, -apple-system, system-ui, sans-serif", font_color="#a8bcd0", margin=dict(l=36, r=16, t=44, b=36), hoverlabel=dict( bgcolor="rgba(6,8,13,0.97)", bordercolor="rgba(255,255,255,0.12)", font_size=11, font_family="Inter, sans-serif", font_color="#e8edf5", ), ) # Axis presets _AXIS_Y = dict( gridcolor="rgba(255,255,255,0.04)", zerolinecolor="rgba(255,255,255,0.06)", showline=False, tickfont=dict(size=10, color="#6a8099"), ) _AXIS_X = dict( gridcolor="rgba(0,0,0,0)", zerolinecolor="rgba(0,0,0,0)", showline=False, tickfont=dict(size=10, color="#6a8099"), ) _AXIS_NONE = dict(showgrid=False, zeroline=False, showline=False, tickfont=dict(size=10, color="#6a8099")) # ───────────────────────────────────────────────────────────── # Chart Builders — pure functions, no side-effects # ───────────────────────────────────────────────────────────── def _chart_velocity(signals: list[Signal]) -> go.Figure: """Stacked area: signal ingest per day per dimension, last 30 days.""" now = datetime.now(timezone.utc) day_dim: dict[str, list[int]] = {d: [0] * 30 for d in _DIM_COLOUR} for s in signals: ts = s.date_ingested if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) delta = (now - ts).days if 0 <= delta < 30: day_dim[s.pestel_dimension.value][29 - delta] += 1 fig = go.Figure() for dim, col in _DIM_COLOUR.items(): fig.add_trace(go.Scatter( x=list(range(30)), y=day_dim[dim], name=dim, mode="lines", stackgroup="one", line=dict(width=0.5, color=col), fillcolor=_hex_to_rgba(col, 0.25), hovertemplate=f"{dim}: %{{y}}", )) fig.update_layout( **_CHART_BASE, title=dict(text=f"Signal Ingest · Last 30 Days ({len(signals)} total)", font_size=11, x=0.5, xanchor="center", y=0.96, font_color="#c4d2de"), xaxis=dict(title="", **_AXIS_X, tickvals=[0, 9, 19, 29], ticktext=["30d ago", "20d ago", "10d ago", "Today"]), yaxis=dict(title="Signals", **_AXIS_Y), legend=dict(orientation="h", y=-0.20, font_size=10, bgcolor="rgba(0,0,0,0)"), height=260, ) return fig def _chart_pestel_bar(signals: list[Signal]) -> go.Figure: """Horizontal bar: avg disruption per dimension.""" dim_scores: dict[str, list[float]] = defaultdict(list) for s in signals: dim_scores[s.pestel_dimension.value].append(s.disruption_score) dims = ["LEGAL", "TECHNOLOGICAL", "POLITICAL", "ECONOMIC", "ENVIRONMENTAL", "SOCIAL"] avgs = [round(sum(dim_scores[d]) / len(dim_scores[d]), 3) if dim_scores[d] else 0.0 for d in dims] cols = [_DIM_COLOUR[d] for d in dims] fig = go.Figure(go.Bar( x=avgs, y=dims, orientation="h", marker=dict(color=cols, opacity=0.80, line=dict(color="rgba(0,0,0,0)", width=0)), hovertemplate="%{y}: %{x:.3f}", )) fig.update_layout( **_CHART_BASE, title=dict(text="Avg Disruption by Dimension", font_size=11, x=0.5, xanchor="center", y=0.96, font_color="#c4d2de"), xaxis=dict(range=[0, 1], title="", **_AXIS_X), yaxis=dict(**_AXIS_X), showlegend=False, height=240, ) return fig def _chart_histogram(signals: list[Signal]) -> go.Figure: """Disruption score distribution — stacked bar by PESTEL dimension.""" # Score buckets: LOW <0.40, MODERATE 0.40–0.60, HIGH 0.60–0.75, CRITICAL ≥0.75 buckets = ["LOW\n<0.40", "MODERATE\n0.40–0.60", "HIGH\n0.60–0.75", "CRITICAL\n≥0.75"] def _bucket(score: float) -> int: if score < 0.40: return 0 if score < 0.60: return 1 if score < 0.75: return 2 return 3 # Count per dimension per bucket counts: dict[str, list[int]] = {d: [0, 0, 0, 0] for d in _DIM_COLOUR} for s in signals: counts[s.pestel_dimension.value][_bucket(s.disruption_score)] += 1 fig = go.Figure() if not signals: fig.add_annotation( text="No signals yet — run the Scout pipeline.", x=0.5, y=0.5, xref="paper", yref="paper", showarrow=False, font=dict(size=12, color="#a8bcd0"), ) else: for dim, col in _DIM_COLOUR.items(): fig.add_trace(go.Bar( name=dim, x=buckets, y=counts[dim], marker_color=_hex_to_rgba(col, 0.82), marker_line_width=0, hovertemplate=f"{dim}: %{{y}} signals", )) fig.update_layout( **_CHART_BASE, barmode="stack", title=dict( text=f"Disruption Distribution · {len(signals)} signals", font_size=11, x=0.5, xanchor="center", y=0.96, font_color="#ffffff", ), xaxis=dict(title="", **_AXIS_X), yaxis=dict(title="Signals", **_AXIS_Y), bargap=0.22, legend=dict(orientation="h", y=-0.22, font_size=9, bgcolor="rgba(0,0,0,0)", font_color="#c4d2de"), height=240, ) return fig def _chart_radar(signals: list[Signal], dim_filter: str = "All", min_score: float = 0.50) -> go.Figure: """ Innovation Radar — disruption score → 12M/24M/36M time ring. Default min_score=0.50 shows only HIGH+CRITICAL (sponsor requirement #4). """ filtered = [ s for s in signals if s.disruption_score >= min_score and (dim_filter == "All" or s.pestel_dimension.value == dim_filter) ] fig = go.Figure() # Ring shading — stronger reds near 12M for x0, x1, fill in [ (0, 12, "rgba(255,23,68,0.08)"), (12, 24, "rgba(255,171,0,0.05)"), (24, 37, "rgba(0,230,118,0.03)"), ]: fig.add_vrect(x0=x0, x1=x1, fillcolor=fill, line_width=0, layer="below") if not filtered: fig.add_annotation( text="No signals at this threshold. Lower the filter or run the Scout.", x=18, y=0.5, showarrow=False, font=dict(size=13, color="#6a8099"), ) else: for dim, col in _DIM_COLOUR.items(): sigs = [s for s in filtered if s.pestel_dimension.value == dim] if not sigs: continue xs, ys, sizes, labels = [], [], [], [] for s in sigs: seed = int(hashlib.md5(s.id.encode()).hexdigest(), 16) % (2 ** 32) rng = np.random.default_rng(seed) j = rng.uniform(-3.0, 3.0) if s.disruption_score >= 0.75: x = float(np.clip(6.5 + j, 2, 11)) elif s.disruption_score >= 0.50: x = float(np.clip(18.0 + j, 13, 23)) else: x = float(np.clip(30.0 + j, 25, 35)) xs.append(x) ys.append(s.disruption_score) sizes.append(max(10, s.disruption_score * 36)) labels.append(s.title[:55] + ("…" if len(s.title) > 55 else "")) fig.add_trace(go.Scatter( x=xs, y=ys, name=dim, mode="markers", marker=dict(size=sizes, color=col, opacity=0.85, line=dict(color="rgba(255,255,255,0.15)", width=1)), text=labels, customdata=[[s.disruption_score, s.pestel_dimension.value, s.source_url] for s in sigs], hovertemplate=( "%{text}
" "Dimension: %{customdata[1]}
" "Disruption: %{customdata[0]:.3f}
" "Source →" "" ), )) # Bold, prominent time-horizon ring lines (sponsor requirement #2) for x, lbl, col in [ (12, "▐ 12M · CRITICAL", "#ff1744"), (24, "▐ 24M · HIGH", "#ffab00"), (36, "▐ 36M · MONITOR", "#00e676"), ]: fig.add_vline( x=x, line=dict(color=col, width=2.5, dash="solid"), annotation_text=lbl, annotation=dict( font_size=10, font_color=col, y=1.04, yref="paper", bgcolor=_hex_to_rgba(col, 0.12), bordercolor=col, borderwidth=1, borderpad=4, ), ) fig.update_layout( **_CHART_BASE, title=dict( text=f"Innovation Radar · {len(filtered)} of {len(signals)} signals · score ≥ {min_score:.2f}", font_size=11, x=0.5, xanchor="center", y=0.97, font_color="#c4d2de", ), xaxis=dict(title="Time to Impact (months)", range=[0, 38], **_AXIS_NONE), yaxis=dict(title="Disruption Score", range=[0, 1.10], **_AXIS_NONE), legend=dict(orientation="v", x=1.01, font_size=10, bgcolor="rgba(0,0,0,0)"), height=480, ) return fig def _build_export_html() -> str: """Build a self-contained HTML report string for download.""" signals = _get_all_signals_cached() stats = _db_stats_cached() top10 = sorted(signals, key=lambda s: s.disruption_score, reverse=True)[:10] critical = [s for s in signals if s.disruption_score >= 0.75] now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") rows = "\n".join( f"" f"{s.pestel_dimension.value}" f"{s.title[:120]}" f"{s.disruption_score:.3f}" f"{_sev(s.disruption_score).upper()}" f"Source" f"" for s in top10 ) critical_rows = "\n".join( f"{s.pestel_dimension.value}{s.title[:120]}" f"{s.disruption_score:.3f}Source" for s in critical[:5] ) return f""" Fendt PESTEL-EL Intelligence Report — {now_str}

Fendt PESTEL-EL Strategic Intelligence Report

Generated: {now_str}  |  EU Data Act 2026 Compliant

{stats['total']}
Total Signals
{stats['critical']}
Critical (≥0.75)
{stats['high']}
High (0.50–0.75)
{stats['avg_disruption']:.3f}
Avg Disruption

Urgency Matrix — 12M Critical Signals

{critical_rows if critical_rows else ''}
DimensionSignalScoreSource
No critical signals at this time.

Top 10 Signals by Disruption Score

{rows if rows else ''}
DimensionSignalScoreSeveritySource
No signals found. Run the Scout to ingest intelligence.

This report was generated by the Fendt PESTEL-EL Sentinel. All signals include verifiable source URLs in compliance with EU Data Act 2026 provenance requirements.

""" # ───────────────────────────────────────────────────────────── # Pre-flight — crash before Dash starts if any builder breaks # ───────────────────────────────────────────────────────────── def _preflight() -> None: try: signals = _get_all_signals_cached() _chart_velocity(signals) _chart_pestel_bar(signals) _chart_histogram(signals) _chart_radar(signals) log.info("Pre-flight passed — all chart builders OK (%d signals)", len(signals)) except Exception as exc: log.warning("Pre-flight skipped or failed (likely DB connection): %s", exc) # ───────────────────────────────────────────────────────────── # Micro-components (CSS className throughout) # ───────────────────────────────────────────────────────────── def _metric(label: str, value: str, sub: str = "", glow: str = "") -> html.Div: cls = f"kpi-card{(' glow-' + glow) if glow else ''}" return html.Div([ html.Div(label, className="kpi-label"), html.Div(value, className="kpi-value"), html.Div(sub, className="kpi-sub") if sub else html.Span(), ], className=cls) def _dot(label: str, kind: str = "idle") -> html.Div: return html.Div([ html.Div(className=f"sb-dot dot-{kind}"), html.Span(label), ], className="sb-status") # ───────────────────────────────────────────────────────────── # Urgency Matrix — 12M CRITICAL signals # ───────────────────────────────────────────────────────────── def _urgency_card(s: Signal) -> html.Div: dim_code = _DIM_PILL_CODE.get(s.pestel_dimension.value, "P") return html.Div([ html.Span(s.pestel_dimension.value[:3], className=f"dim-pill dp-{dim_code}"), html.Div(f"{s.disruption_score:.3f}", className="urgency-score"), html.Div(s.title, className="urgency-title"), html.A("↗ Verify Source", href=s.source_url, target="_blank", className="source-link"), ], className="urgency-card") def _urgency_matrix(signals: list[Signal]) -> html.Div: critical = sorted( [s for s in signals if s.disruption_score >= 0.75], key=lambda s: s.disruption_score, reverse=True, )[:3] if not critical: return html.Div([ html.Div("URGENCY MATRIX — 12M CRITICAL", className="section-label"), html.Div( "No critical signals (score ≥ 0.75). Run the Scout to ingest intelligence.", style={"fontSize": "12px", "color": "#e8edf5"}, ), ], className="mb-4") return html.Div([ html.Div("URGENCY MATRIX — 12M CRITICAL SIGNALS", className="section-label"), dbc.Row( [dbc.Col(_urgency_card(s), md=4) for s in critical], className="g-3", ), ], className="mb-4") # ───────────────────────────────────────────────────────────── # Tab Content Builders # ───────────────────────────────────────────────────────────── def _tab_overview() -> html.Div: """Field Intelligence Overview — KPIs and high-level distribution.""" signals = _get_unique_signals_cached() stats = _db_stats_cached() top3 = sorted(signals, key=lambda s: s.disruption_score, reverse=True)[:3] return html.Div([ # KPI Row dbc.Row([ dbc.Col(_metric("Total Signals", str(stats["total"]), "Astra DB Vector Store"), md=3), dbc.Col(_metric("Critical", str(stats["critical"]), "Score ≥ 0.75", "red"), md=3), dbc.Col(_metric("High", str(stats["high"]), "Score 0.50–0.75", "amber"), md=3), dbc.Col(_metric("Avg Disruption", f"{stats['avg_disruption']:.3f}", "Global Mean"), md=3), ], className="g-3 mb-3"), # Charts Row dbc.Row([ dbc.Col(html.Div(dcc.Graph(figure=_chart_velocity(signals), id="chart-velocity", config={"displayModeBar": False}), className="chart-card"), md=7), dbc.Col(html.Div(dcc.Graph(figure=_chart_pestel_bar(signals), id="chart-pestel-bar", config={"displayModeBar": False}), className="chart-card"), md=5), ], className="g-3 mb-3"), dbc.Row([ dbc.Col(html.Div( dcc.Graph(figure=_chart_histogram(signals), config={"displayModeBar": False}), className="chart-card", ), md=6), dbc.Col(html.Div([ html.Div("TOP SIGNALS — HIGH + CRITICAL", className="section-label"), *([_urgency_card(s) for s in top3] if top3 else [ html.P("No signals yet. Run the Scout.", style={"color": "#e8edf5", "fontSize": "12px"}), ]), ], className="war-card"), md=6), ], className="g-3"), ]) def _tab_radar() -> html.Div: """Disruption Horizon — innovation radar view.""" return html.Div([ dbc.Row([ dbc.Col(html.Div([ # Ring guide html.Div("RING GUIDE", className="section-label"), *[html.Div([ html.Div(style={"width": "10px", "height": "10px", "borderRadius": "50%", "background": col, "flexShrink": "0", "boxShadow": f"0 0 6px {col}"}), html.Span(f"{ring} — {lbl}", style={"fontSize": "11px", "color": "#ffffff"}), ], style={"display": "flex", "alignItems": "center", "gap": "8px", "marginBottom": "7px"}) for ring, lbl, col in [ ("12M", "CRITICAL · Immediate", "#ff1744"), ("24M", "HIGH · Plan Now", "#ffab00"), ("36M", "MONITOR · Watch Horizon", "#00e676"), ]], html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "16px 0"}), # Filters html.Div("FILTERS", className="section-label"), html.Div("PESTEL Dimension", className="filter-label"), dcc.Dropdown( id="radar-dim-filter", options=[{"label": d, "value": d} for d in ["All", "POLITICAL", "ECONOMIC", "SOCIAL", "TECHNOLOGICAL", "ENVIRONMENTAL", "LEGAL"]], value="All", clearable=False, className="dark-dropdown", style={"marginBottom": "14px"}, ), html.Div("Min Disruption Score", className="filter-label"), dcc.Slider( id="radar-score-slider", min=0, max=1, step=0.05, value=0.50, marks={0: "0", 0.50: "0.50", 0.75: "0.75", 1: "1"}, tooltip={"placement": "bottom", "always_visible": False}, className="dark-slider", ), html.Div( "Default shows HIGH + CRITICAL only (≥ 0.50)", style={"fontSize": "9.5px", "color": "#ffffff", "marginTop": "8px"}, ), ], className="war-card"), md=3), dbc.Col(html.Div([ dcc.Graph(id="radar-chart", config={"displayModeBar": False}), ], className="chart-card"), md=9), ], className="g-3"), dbc.Row([ dbc.Col(html.Div(id="radar-table-container", style={"marginTop": "20px"}), md=12), ], className="g-3"), ]) def _row(s: Signal) -> html.Tr: """Module-level row builder for the Signal Feed table.""" return html.Tr([ html.Td(s.date_ingested.strftime("%Y-%m-%d")), html.Td(html.Span(_DIM_PILL_CODE.get(s.pestel_dimension.value, "?"), className=f"dim-pill dp-{_DIM_PILL_CODE.get(s.pestel_dimension.value, 'P')}")), html.Td(html.Div([ html.Div(s.title, style={"fontWeight": "600", "color": "#e8edf5"}), html.Div(s.content[:140] + "...", style={"fontSize": "11px", "color": "#7d8fa8"}), ])), html.Td(f"{s.disruption_score:.3f}", style={"fontFamily": "JetBrains Mono", "color": _SEV_COLOUR.get(_sev(s.disruption_score))}), html.Td(html.A("↗", href=s.source_url, target="_blank", className="source-link")), ]) def _tab_feed() -> html.Div: """Signal Feed — raw intelligence data table.""" stats = _db_stats_cached() by_dim = stats.get("by_dimension", {}) return html.Div([ dbc.Row([ dbc.Col([ html.Div([ dcc.Dropdown( id="feed-sort-dropdown", options=[ {"label": "Newest First", "value": "newest"}, {"label": "Highest Disruption", "value": "score_desc"}, {"label": "Lowest Disruption", "value": "score_asc"}, ], value="newest", clearable=False, className="dark-dropdown", style={"width": "200px"}, ), dcc.Dropdown( id="feed-dim-dropdown", options=[{"label": "All Dimensions", "value": "ALL"}] + [{"label": d, "value": d} for d in ["POLITICAL", "ECONOMIC", "SOCIAL", "TECHNOLOGICAL", "ENVIRONMENTAL", "LEGAL"]], value="ALL", clearable=False, className="dark-dropdown", style={"width": "220px"}, ), ], style={"display": "flex", "gap": "12px", "marginBottom": "14px", "flexWrap": "wrap"}), html.Div( id="feed-count-label", style={"fontSize": "11px", "color": "#e8edf5", "marginBottom": "16px"}, ), html.Table([ html.Thead(html.Tr([ html.Th("Date"), html.Th("Dim"), html.Th("Signal"), html.Th("Score"), html.Th("Src"), ])), html.Tbody([], id="feed-table-body"), ], className="war-table"), ], md=8), dbc.Col(html.Div([ html.Div("DATABASE", className="section-label"), _metric("Total", str(stats["total"]) if stats["total"] else "—"), html.Div(style={"height": "10px"}), _metric("Critical", str(stats["critical"]) if stats["total"] else "—", glow="red"), html.Div(style={"height": "10px"}), _metric("Avg Score", f'{stats["avg_disruption"]:.3f}' if stats["total"] else "—"), html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "16px 0"}), html.Div("BY DIMENSION", className="section-label"), *[html.Div([ html.Span(d[:3], style={"fontSize": "10px", "fontWeight": "600", "color": _DIM_COLOUR.get(d, "#7d8fa8"), "minWidth": "36px", "display": "inline-block"}), html.Span(str(by_dim.get(d, 0)), style={"fontFamily": "JetBrains Mono, monospace", "fontSize": "11px", "color": "#e8edf5"}), ], style={"marginBottom": "6px"}) for d in ["POLITICAL", "ECONOMIC", "SOCIAL", "TECHNOLOGICAL", "ENVIRONMENTAL", "LEGAL"]], ], className="war-card"), md=4), ], className="g-3"), ]) def _chat_bubble(text: str, role: str = "assistant") -> html.Div: cls = f"chat-bubble bubble-{role}" return html.Div([ html.Div(textwrap.fill(text, 100) if role == "user" else dcc.Markdown(text), className=cls), ], style={"display": "flex", "justifyContent": "flex-end" if role == "user" else "flex-start", "marginBottom": "12px"}) def _tab_chatbot(history: list) -> html.Div: """Strategic Advisor — conversational AI interface.""" welcome = _chat_bubble( "**Fendt Commercial Intelligence Advisor**\n\n" "I'm your embedded Fendt/AGCO marketing & sales analyst. " "Ask me about dealer positioning, competitive differentiation against Deere/CNH/Claas, " "precision farming upsell narratives, regulatory urgency messaging, or pipeline priorities " "— all grounded in live PESTEL signals from the database.\n\n" "Questions are routed automatically: data questions go to the **Calculator Agent**, " "strategic and messaging questions go to the **Analyst Agent**.", role="assistant", ) bubbles = [welcome] for msg in history: bubble = _chat_bubble(msg["text"], msg["role"]) if msg["role"] == "assistant" and msg.get("badge"): badge = html.Div( msg["badge"], style={ "fontSize": "9px", "fontFamily": "JetBrains Mono, monospace", "color": msg.get("badge_colour", "#7d8fa8"), "marginTop": "6px", "opacity": "0.75", }, ) bubble = html.Div([bubble, badge]) bubbles.append(bubble) chips = [ "Which signals should Fendt's sales team lead with in dealer conversations this quarter?", "How should Fendt marketing position the Vario tractor line against CNH and Deere given current EU signals?", "What precision farming trends give AGCO the strongest upsell narrative to existing customers?", "Which regulatory changes create urgency for farmers to upgrade equipment — and how do we message that?", "What competitive threats from John Deere, CNH, or Claas should Fendt sales reps be prepared to counter?", ] return html.Div([ dbc.Row([ # ── Left: Chat Window ────────────────────────────────── dbc.Col([ html.Div(bubbles, id="chat-messages", className="chat-window"), html.Div([ dcc.Input(id="chat-input", placeholder="Ask a strategic question...", className="chat-input-field", n_submit=0), dbc.Button("Send", id="chat-send", color="primary", className="chat-btn"), ], className="chat-input-group"), ], md=7), # ── Right: Try Questions + Agent Info ────────────────── dbc.Col(html.Div([ # Strategic prompts html.Div("STRATEGIC PROMPTS", className="section-label"), *[html.Button( c, id=f"chip-{i}", n_clicks=0, className="advisor-chip", ) for i, c in enumerate(chips)], html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "18px 0"}), # Agent capabilities html.Div("AGENT CAPABILITIES", className="section-label"), html.Div([ html.Div([ html.Div("◆", style={"color": "#00e5ff", "fontSize": "8px", "marginRight": "8px", "marginTop": "2px"}), html.Div([ html.Div("Calculator Agent", style={"color": "#00e5ff", "fontSize": "11px", "fontWeight": "600"}), html.Div("Signal scoring, pipeline metrics, disruption rankings by dimension.", style={"fontSize": "10px", "color": "#7d8fa8", "marginTop": "2px"}), ]), ], style={"display": "flex", "marginBottom": "12px"}), html.Div([ html.Div("◆", style={"color": "#ffd93d", "fontSize": "8px", "marginRight": "8px", "marginTop": "2px"}), html.Div([ html.Div("Analyst Agent", style={"color": "#ffd93d", "fontSize": "11px", "fontWeight": "600"}), html.Div("Sales positioning, competitive messaging, marketing actions vs Deere/CNH/Claas.", style={"fontSize": "10px", "color": "#7d8fa8", "marginTop": "2px"}), ]), ], style={"display": "flex"}), ]), html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "18px 0"}), # Coverage areas html.Div("KNOWLEDGE AREAS", className="section-label"), *[html.Div(area, style={"fontSize": "10px", "color": "#7d8fa8", "marginBottom": "4px", "paddingLeft": "8px", "borderLeft": "2px solid rgba(0,229,255,0.2)"}) for area in [ "Fendt Vario / IDEAL / FendtONE portfolio", "AGCO · Massey Ferguson · Valtra · Challenger", "Competitive: Deere, CNH, Claas, Kubota", "EU CAP reform & subsidy dynamics", "Precision ag adoption & dealer economics", "Emissions regs · electrification pipeline", "Eastern EU market expansion signals", "Commodity price → farmer buying intent", ]], html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "18px 0"}), # Model info html.Div("MODEL", className="section-label"), html.Div(_HF_REPO_ID, style={"fontFamily": "JetBrains Mono, monospace", "fontSize": "10px", "color": "#e8edf5"}), html.Div("Fendt/AGCO Commercial Intelligence", style={"fontSize": "10px", "color": "#7d8fa8", "marginTop": "4px"}), ], className="war-card"), md=5), ], className="g-3"), ]) # ───────────────────────────────────────────────────────────── # Knowledge Graph components # ───────────────────────────────────────────────────────────── _GRAPH_JSON = Path(__file__).parent / "data" / "graph.json" _CAT_COLOUR = { "POLITICAL": "#64b5f6", "ECONOMIC": "#a5d6a7", "SOCIAL": "#ffcc80", "TECHNOLOGICAL": "#ce93d8", "ENVIRONMENTAL": "#80deea", "LEGAL": "#ef9a9a", } _CYTO_STYLESHEET = [ { "selector": "node", "style": { "label": "data(label)", "background-color": "data(colour)", "color": "#ffffff", "font-size": "11px", "font-weight": "600", "font-family": "Inter, -apple-system, sans-serif", "text-wrap": "wrap", "text-max-width": "130px", "text-valign": "bottom", "text-margin-y": "6px", "text-background-color": "rgba(6,8,13,0.80)", "text-background-opacity": "1", "text-background-padding": "3px", "width": "34px", "height": "34px", "border-width": "2px", "border-color": "rgba(255,255,255,0.25)", "box-shadow": "0 0 8px data(colour)", }, }, { "selector": "edge", "style": { "line-color": "#4fc3f7", "target-arrow-color": "#4fc3f7", "target-arrow-shape": "triangle", "arrow-scale": "1.4", "curve-style": "bezier", "opacity": "0.9", "width": "data(weight_px)", "label": "data(relationship)", "font-size": "10px", "font-family": "Inter, sans-serif", "color": "#e8edf5", "text-opacity": "1", "text-rotation": "autorotate", "text-background-color": "rgba(6,8,13,0.85)", "text-background-opacity": "1", "text-background-padding": "3px", }, }, { "selector": "node:selected", "style": { "border-color": "#00e5ff", "border-width": "3px", "box-shadow": "0 0 14px #00e5ff", }, }, { "selector": "edge:selected", "style": { "line-color": "#00e5ff", "target-arrow-color": "#00e5ff", "opacity": "1", "width": "3", }, }, ] def _load_graph_elements() -> list[dict]: """Load data/graph.json and convert to cytoscape elements format. Only nodes that participate in at least one edge are included — isolated nodes add visual noise without conveying relationships. """ if not _GRAPH_JSON.exists(): return [] try: raw = json.loads(_GRAPH_JSON.read_text()) # Only include nodes that participate in at least one edge connected_ids: set[str] = set() for link in raw.get("links", []): connected_ids.add(link["source"]) connected_ids.add(link["target"]) elements: list[dict] = [] for node in raw.get("nodes", []): if node["id"] not in connected_ids: continue cat = node.get("category", "") elements.append({ "data": { "id": node["id"], "label": node.get("label", node["id"])[:40], "colour": _CAT_COLOUR.get(cat, "#9cb3c9"), "category": cat, }, }) for link in raw.get("links", []): weight = link.get("weight", 0.5) elements.append({ "data": { "source": link["source"], "target": link["target"], "relationship": link.get("relationship", ""), "weight_px": max(2, int(weight * 6)), }, }) return elements except Exception as exc: log.warning("_load_graph_elements failed: %s", exc) return [] def _render_causal_chains() -> list: """Build sidebar widgets for the top causal cascade chains.""" try: chains = get_causal_chains(top_n=5) except Exception: chains = [] if not chains: return [html.Div( "No cascade chains yet — chains build as signals relate to each other.", style={"fontSize": "9px", "color": "#e8edf5", "lineHeight": "1.6"}, )] items = [] for c in chains: chain_parts = c["chain"] arrow_chain_nodes: list = [] for i, p in enumerate(chain_parts): arrow_chain_nodes.append( html.Span(p[:3], style={"color": _CAT_COLOUR.get(p, "#7d8fa8")}) ) if i < len(chain_parts) - 1: arrow_chain_nodes.append(" → ") items.append(html.Div([ html.Div( f"depth {c['depth']} · {c['predicate']}", style={"fontSize": "9px", "color": "#e8edf5", "fontFamily": "JetBrains Mono, monospace"}, ), html.Div( arrow_chain_nodes, style={"fontSize": "10px", "marginTop": "2px"}, ), ], style={"marginBottom": "8px", "paddingLeft": "4px", "borderLeft": "2px solid rgba(0,229,255,0.3)"})) return items def _render_inferred_relationships() -> list: """Build sidebar widgets for inferred cross-PESTEL cascade relationships.""" if not _GRAPH_JSON.exists(): return [] try: graph = json.loads(_GRAPH_JSON.read_text()) inferred = [ t for t in graph.get("triples", []) if t.get("metadata", {}).get("inferred") ] except Exception as exc: log.warning("_render_inferred_relationships: %s", exc) return [] if not inferred: return [html.Div( "No inferred cascades yet — click 'Run Inference' to surface hidden cross-PESTEL relationships.", style={"fontSize": "9px", "color": "#e8edf5", "lineHeight": "1.6"}, )] items = [] for t in inferred[:5]: chain = t.get("metadata", {}).get("causal_chain", []) hops = t.get("metadata", {}).get("hop_count", 0) subj = t.get("subject", {}).get("label", "?") obj = t.get("object", {}).get("label", "?") arrow_nodes: list = [] for i, p in enumerate(chain): arrow_nodes.append(html.Span(p[:3], style={"color": _CAT_COLOUR.get(p, "#7d8fa8")})) if i < len(chain) - 1: arrow_nodes.append(" → ") items.append(html.Div([ html.Div( f"{hops}-hop cascade", style={"fontSize": "9px", "color": "#00e5ff", "fontFamily": "JetBrains Mono, monospace"}, ), html.Div( arrow_nodes, style={"fontSize": "10px", "marginTop": "2px"}, ), html.Div( f"{subj[:28]} → {obj[:28]}", style={"fontSize": "9px", "color": "#e8edf5", "marginTop": "2px"}, ), ], style={"marginBottom": "8px", "paddingLeft": "4px", "borderLeft": "2px solid rgba(0,229,255,0.15)"})) return items def _tab_graph(status: str = "") -> html.Div: """Knowledge Graph — causal interdependency visualisation.""" elements = _load_graph_elements_cached() node_count = sum(1 for e in elements if "source" not in e.get("data", {})) edge_count = len(elements) - node_count has_data = node_count > 0 graph_controls = html.Div([ dbc.Button( "Rebuild Graph", id="rebuild-graph-btn", color="warning", size="sm", outline=True, style={"marginRight": "8px", "fontSize": "10px"}, ), dbc.Button( "Run Inference", id="run-inference-btn", color="info", size="sm", outline=True, style={"fontSize": "10px"}, ), html.Div(id="graph-action-status", children=status, style={"fontSize": "10px", "color": "#e8edf5", "marginTop": "6px"}), ], style={"marginBottom": "12px"}) legend = [ html.Div([ html.Div(style={"width": "10px", "height": "10px", "borderRadius": "50%", "background": col, "flexShrink": "0", "boxShadow": f"0 0 5px {col}"}), html.Span(cat, style={"fontSize": "10px", "color": "#e8edf5"}), ], style={"display": "flex", "alignItems": "center", "gap": "8px", "marginBottom": "6px"}) for cat, col in _CAT_COLOUR.items() ] # Cytoscape is always rendered so the callback target always exists. # Empty state is an absolute overlay that disappears once nodes arrive. graph_canvas = html.Div([ cyto.Cytoscape( id="knowledge-graph", elements=elements, layout={ "name": "cose", "animate": False, "nodeRepulsion": 8000, "idealEdgeLength": 140, "gravity": 0.03, "numIter": 2500, "padding": 50, "componentSpacing": 100, "nodeDimensionsIncludeLabels": True, "randomize": True, }, stylesheet=_CYTO_STYLESHEET, style={"width": "100%", "height": "580px", "background": "rgba(13,17,23,0.95)", "borderRadius": "8px"}, ), # Empty-state overlay — covers graph area when there are no nodes html.Div([ html.Div("○", className="empty-state-icon"), html.Div("No graph data yet", className="empty-state-title"), html.Div( "Run the Scout to ingest signals. The Knowledge Graph builds automatically " "after each cycle. Use 'Rebuild Graph' to reconstruct from the current DB state.", className="empty-state-body", ), ], className="empty-state", style={ "display": "none" if has_data else "flex", "position": "absolute", "top": "0", "left": "0", "right": "0", "bottom": "0", "borderRadius": "8px", "background": "rgba(13,17,23,0.95)", "zIndex": "10", }), ], style={"position": "relative"}) return html.Div([ dbc.Row([ dbc.Col(html.Div(graph_canvas, className="chart-card"), md=9), dbc.Col(html.Div([ graph_controls, html.Div("GRAPH INFO", className="section-label"), _metric("Nodes", str(node_count) if has_data else "—"), html.Div(style={"height": "8px"}), _metric("Edges", str(edge_count) if has_data else "—"), html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "14px 0"}), html.Div("INFERRED CASCADES", className="section-label"), *_render_inferred_relationships(), html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "14px 0"}), html.Div("CAUSAL CHAINS", className="section-label"), *_render_causal_chains(), html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "14px 0"}), html.Div("DIMENSION KEY", className="section-label"), *legend, html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "14px 0"}), html.Div("Click a node to inspect · Drag to rearrange", style={"fontSize": "10px", "color": "#a8bcd0", "marginTop": "6px"}), ], className="war-card"), md=3), ], className="g-3"), ]) # ── Reports helpers ─────────────────────────────────────────── _REPORTS_DIR = Path(__file__).parent / "outputs" / "reports" def _glob_reports() -> list[dict]: """Return sorted list of {label, value} dicts for available .md reports.""" try: if not _REPORTS_DIR.exists(): return [] paths = sorted(_REPORTS_DIR.glob("*.md"), key=lambda p: p.stat().st_mtime, reverse=True) return [{"label": p.stem.replace("_", " ").title(), "value": str(p)} for p in paths] except Exception: return [] def _render_report_body(path: str | None) -> html.Div: """Build the full styled report viewer for a given .md path.""" if not path: return html.Div([ html.Div("📄", className="empty-state-icon"), html.Div("No reports available", className="empty-state-title"), html.Div( "Generate a report by running the Sentinel pipeline, then place the .md file " "in outputs/reports/ to register it here.", className="empty-state-body", ), ], className="empty-state") try: content = Path(path).read_text(encoding="utf-8") except OSError as exc: log.error("_render_report_body: cannot read %s: %s", path, exc) return html.P(f"Could not read report: {exc}", style={"color": "#ff6090", "fontSize": "12px", "padding": "16px"}) # Extract title and generated date from first lines if present lines = content.splitlines() doc_title = lines[0].lstrip("# ").strip() if lines else Path(path).stem doc_date = "" doc_class = "CONFIDENTIAL — C-SUITE ONLY" for line in lines[1:6]: if line.startswith("**Generated:**"): doc_date = line.replace("**Generated:**", "").strip() if line.startswith("**Classification:**"): doc_class = line.replace("**Classification:**", "").strip() # Strip first heading + metadata from content before rendering body body_start = 0 for i, ln in enumerate(lines): if i > 0 and ln.startswith("---"): body_start = i + 1 break body_md = "\n".join(lines[body_start:]) if body_start else content return html.Div([ # ── Document header ────────────────────────────────── html.Div([ html.Div([ html.Div(doc_class, className="report-classification"), html.Div(doc_title, className="report-title"), html.Div([ html.Span("Generated: ", style={"color": "#e8edf5"}), html.Span(doc_date or "—", style={"color": "#e8edf5"}), html.Span(" · Source: Fendt PESTEL-EL Sentinel", style={"color": "#e8edf5"}), ], className="report-meta"), ], className="report-doc-header-left"), ], className="report-doc-header"), # ── Report body ─────────────────────────────────────── dcc.Markdown(body_md, dangerously_allow_html=True, className="report-markdown"), ]) def _sanitize_for_pdf(text: str) -> str: """Translate non-Latin-1 characters that fpdf Helvetica can't render. fpdf2's built-in fonts are Latin-1 only. Rather than bundle a TTF, we map the most common Unicode glyphs to ASCII equivalents so the export never raises a UnicodeEncodeError. """ _MAP = { "\u20ac": "EUR", # € "\u00a3": "GBP", # £ (already Latin-1, but keep for completeness) "\u2013": "-", # – en-dash "\u2014": "--", # — em-dash "\u2018": "'", # ' left single quote "\u2019": "'", # ' right single quote / apostrophe "\u201c": '"', # " left double quote "\u201d": '"', # " right double quote "\u2026": "...", # … ellipsis "\u00b7": "*", # · middle dot "\u00a0": " ", # non-breaking space "\u2022": "*", # • bullet "\u25cf": "*", # ● filled circle "\u2192": "->", # → right arrow "\u2190": "<-", # ← left arrow } for char, replacement in _MAP.items(): text = text.replace(char, replacement) # Drop any remaining non-Latin-1 characters silently return text.encode("latin-1", errors="ignore").decode("latin-1") def _md_to_pdf_bytes(content: str) -> bytes: """Convert markdown content to a PDF byte string using fpdf2.""" from fpdf import FPDF # type: ignore[import] import markdown as md_lib # type: ignore[import] html_body = md_lib.markdown(_sanitize_for_pdf(content), extensions=["tables", "fenced_code"]) # fpdf2's write_html understands / not / html_body = ( html_body .replace("", "").replace("", "") .replace("", "").replace("", "") .replace("
", "

").replace("

", "

") .replace("", "").replace("", "") .replace("
", "
").replace("
", "

") .replace("", "").replace("", "") ) pdf = FPDF() pdf.set_margins(25, 22, 25) pdf.set_auto_page_break(auto=True, margin=20) pdf.add_page() pdf.set_font("Helvetica", size=11) pdf.write_html(html_body) return bytes(pdf.output()) def _tab_reports() -> html.Div: """Strategic Reports — Markdown viewer for AI-generated C-Suite briefs.""" options = _glob_reports() default = options[0]["value"] if options else None initial_body = _render_report_body(default) return html.Div([ dbc.Row([ dbc.Col([ # ── Toolbar ──────────────────────────────────── html.Div([ html.Div([ html.Div("SELECT REPORT", className="section-label", style={"marginBottom": "6px"}), dcc.Dropdown( id="reports-dropdown", options=options, value=default, clearable=False, placeholder="No reports found in outputs/reports/", className="dark-dropdown", ), ], style={"flex": "1"}), dbc.Button( "✦ Generate New Intelligence Brief", id="reports-gen-btn", color="primary", size="sm", className="btn-refresh", style={"alignSelf": "flex-end", "whiteSpace": "nowrap"}, ), dbc.Button( "⬇ Export PDF", id="reports-export-pdf-btn", color="secondary", size="sm", outline=True, className="btn-refresh", style={"alignSelf": "flex-end", "whiteSpace": "nowrap", "opacity": "1" if _PDF_OK else "0.35"}, disabled=not _PDF_OK, ), ], style={"display": "flex", "gap": "16px", "alignItems": "flex-end", "marginBottom": "12px"}), # ── Generation status ─────────────────────────── html.Div("", id="reports-gen-status", style={"fontSize": "11px", "color": "#e8edf5", "marginBottom": "16px", "minHeight": "18px"}), # ── Report body ───────────────────────────────── dcc.Loading( html.Div(initial_body, id="reports-body", className="war-card"), id="reports-body-loading", type="circle", color="#00e5ff", ), ], md=10), dbc.Col(html.Div([ html.Div("REPORTS", className="section-label"), _metric("Available", str(len(options))), html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "14px 0"}), html.Div("PDF EXPORT", className="section-label"), html.Div( "PDF export ready" if _PDF_OK else "Install fpdf2 + markdown to enable PDF export", style={"fontSize": "9px", "color": "#00e676" if _PDF_OK else "#6a8099", "lineHeight": "1.6"}, ), html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "14px 0"}), html.Div( "Place .md files in outputs/reports/ to register them here.", style={"fontSize": "9px", "color": "#e8edf5", "lineHeight": "1.6"}, ), ], className="war-card"), md=2), ], className="g-3"), ]) # ── Intelligence Lens helpers ───────────────────────────────── _LENS_PRESETS = [ "CAP Reform", "Electric Tractor Adoption", "Precision Farming Regulation", "Grain Price Volatility", "EU Green Deal Agriculture", "Right to Repair Policy", "Labour Shortages in Farming", "Custom Search…", ] def _lens_signal_card(sig: "Signal", score: float) -> html.Div: # type: ignore[name-defined] """Render a single signal result card for the Intelligence Lens.""" relevance_pct = f"{score * 100:.0f}%" dim_col = _DIM_COLOUR.get(sig.pestel_dimension.value, "#9cb3c9") bar_w = max(4, int(score * 100)) return html.Div([ html.Div([ html.Span(sig.pestel_dimension.value[:3], style={"color": dim_col, "fontWeight": "700", "fontSize": "10px", "minWidth": "36px"}), html.Span(sig.title, style={"color": "#e8edf5", "fontSize": "12.5px", "fontWeight": "600", "flex": "1", "lineHeight": "1.45"}), html.Span(f"{relevance_pct}", style={"color": dim_col, "fontSize": "10px", "fontFamily": "JetBrains Mono, monospace", "fontWeight": "700", "whiteSpace": "nowrap"}), ], style={"display": "flex", "gap": "10px", "alignItems": "flex-start", "marginBottom": "8px"}), # Relevance bar html.Div(html.Div(style={ "height": "2px", "width": f"{bar_w}%", "background": dim_col, "borderRadius": "1px", "boxShadow": f"0 0 6px {dim_col}", }), style={"background": "rgba(255,255,255,0.06)", "borderRadius": "1px", "marginBottom": "10px", "marginLeft": "46px"}), html.P(sig.content[:220] + ("…" if len(sig.content) > 220 else ""), style={"fontSize": "12px", "color": "#c4d0dc", "margin": "0 0 8px 46px", "lineHeight": "1.7"}), html.A( "↗ verify source", href=sig.source_url, target="_blank", className="source-link", style={"marginLeft": "46px"}, ), ], className="war-card", style={"marginBottom": "10px"}) def _run_lens_search(topic: str | None, custom: str | None = None) -> html.Div: """Execute a semantic search and return result cards (or empty state).""" is_custom = topic == "Custom Search…" query = (custom or "").strip() if is_custom else (topic or "").strip() if not query: return html.Div([ html.Div("🔍", className="empty-state-icon"), html.Div("Enter a search query", className="empty-state-title"), html.Div("Select a macro-trend topic above or type a custom query.", className="empty-state-body"), ], className="empty-state") try: total_signals = len(_get_all_signals_cached()) if total_signals == 0: return html.Div([ html.Div("○", className="empty-state-icon"), html.Div("No signals in Astra DB", className="empty-state-title"), html.Div( 'Click "Run Scout Now" in the sidebar to ingest intelligence. ' "The Intelligence Lens will populate after the first scout cycle completes.", className="empty-state-body", ), ], className="empty-state") results = _get_db().search(query, n_results=5) except Exception as exc: log.error("_run_lens_search crashed: %s", exc) return html.P(f"Search error: {exc}", style={"color": "#ff6090", "fontSize": "12px"}) if not results: return html.Div([ html.Div("○", className="empty-state-icon"), html.Div(f'No matches for "{query}"', className="empty-state-title"), html.Div("Try a broader query or run the Scout to ingest more signals.", className="empty-state-body"), ], className="empty-state") header = html.Div( f'{len(results)} signal(s) matched · query: "{query}"', style={"fontSize": "10px", "color": "#e8edf5", "fontFamily": "JetBrains Mono, monospace", "marginBottom": "14px"}, ) return html.Div([header, *[_lens_signal_card(sig, score) for sig, score in results]]) def _tab_lens() -> html.Div: """Strategic Intelligence Lens — semantic deep-dive via Astra DB.""" initial_results = _run_lens_search(_LENS_PRESETS[0]) return html.Div([ dbc.Row([ dbc.Col([ html.Div("MACRO-TREND TOPIC", className="section-label"), dcc.Dropdown( id="lens-topic-dropdown", options=[{"label": t, "value": t} for t in _LENS_PRESETS], value=_LENS_PRESETS[0], clearable=False, className="dark-dropdown", style={"marginBottom": "10px"}, ), dcc.Input( id="lens-custom-input", type="text", placeholder="Type a custom query (active when 'Custom Search…' selected)…", debounce=True, n_submit=0, className="chat-input-field", style={"marginBottom": "16px", "width": "100%"}, ), dcc.Loading( html.Div(initial_results, id="lens-results"), type="circle", color="#00e5ff", ), ], md=9), dbc.Col(html.Div([ html.Div("HOW IT WORKS", className="section-label"), html.P( "Astra DB semantic search surfaces the most relevant signals for any " "macro-trend query. Results are ranked by cosine similarity using the " "all-MiniLM-L6-v2 embedding model.", style={"fontSize": "10.5px", "color": "#c4d0dc", "lineHeight": "1.7"}, ), html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "14px 0"}), html.Div("TOP-K", className="section-label"), html.Div("5 signals per query", style={"fontFamily": "JetBrains Mono, monospace", "fontSize": "10px", "color": "#e8edf5"}), html.Hr(style={"borderColor": "rgba(255,255,255,0.07)", "margin": "14px 0"}), html.Div("TIP", className="section-label"), html.Div('Select "Custom Search…" and type any free-form topic.', style={"fontSize": "9.5px", "color": "#e8edf5", "lineHeight": "1.6"}), ], className="war-card"), md=3), ], className="g-3"), ]) # ───────────────────────────────────────────────────────────── # Strategic Chat — delegated to Multi-Agent Relational Brain # (core/agents.py: Router → Calculator Agent | Analyst Agent) # ───────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────── # App + Layout # ───────────────────────────────────────────────────────────── # Background-callback cache (DiskCache — survives hot-reload) _CACHE_DIR = Path(__file__).parent / "data" / ".dash_cache" _CACHE_DIR.mkdir(parents=True, exist_ok=True) _disk_cache = diskcache.Cache(str(_CACHE_DIR)) _background_manager = DiskcacheManager(_disk_cache) app = dash.Dash( __name__, external_stylesheets=[ dbc.themes.CYBORG, "https://fonts.googleapis.com/css2?family=Inter:wght@300;400;500;600;700" "&family=JetBrains+Mono:wght@400;600&display=swap", ], suppress_callback_exceptions=True, title="Fendt Sentinel", background_callback_manager=_background_manager, meta_tags=[ {"name": "viewport", "content": "width=device-width, initial-scale=1"}, {"charset": "utf-8"}, ], ) # Bootstrap 5.3 (used by DBC 2.x) requires data-bs-theme="dark" on # for components to render in dark mode. Without this, cards/inputs/dropdowns # render with white backgrounds (Bootstrap's default light theme). app.index_string = """ {%metas%} {%title%} {%favicon%} {%css%} {%app_entry%} """ server = app.server # ── Flask-Caching — memoize expensive DB + graph calls ──────── # SimpleCache keeps results in-process (no Redis needed for single-worker Dash). # 30-second TTL aligns with the auto-refresh interval. _flask_cache = _FlaskCache( app.server, config={"CACHE_TYPE": "SimpleCache", "CACHE_DEFAULT_TIMEOUT": 30}, ) # ── Knowledge-graph rebuild state ───────────────────────────── # Shared between graph_action() (writer) and render_tab() (reader). # SimpleCache is in-process so no cross-process concern; a plain dict is fine. _GRAPH_REBUILD: dict = {"running": False, "status": ""} @_flask_cache.memoize(timeout=30) def _get_all_signals_cached() -> list: """Cached get_all() — avoids hitting Astra DB on every tab render.""" try: return _get_db().get_all() except Exception as exc: log.error("_get_all_signals_cached failed: %s", exc) return [] @_flask_cache.memoize(timeout=30) def _get_unique_signals_cached() -> list: """One signal per source_url — highest disruption_score wins.""" seen: set[str] = set() unique: list = [] for s in sorted(_get_all_signals_cached(), key=lambda s: s.disruption_score, reverse=True): if s.source_url not in seen: seen.add(s.source_url) unique.append(s) return unique @_flask_cache.memoize(timeout=30) def _db_stats_cached() -> dict: """Cached DB stats — avoids a full get_all() on every sidebar tick.""" try: signals = _get_unique_signals_cached() scores = [s.disruption_score for s in signals] by_dim: dict[str, int] = {} for s in signals: dim = s.pestel_dimension.value by_dim[dim] = by_dim.get(dim, 0) + 1 return { "total": len(signals), "critical": sum(1 for sc in scores if sc >= 0.75), "high": sum(1 for sc in scores if 0.50 <= sc < 0.75), "avg_disruption": round(sum(scores) / len(scores), 3) if scores else 0.0, "by_dim": by_dim, "status": "ok", } except Exception as exc: log.warning("_db_stats_cached failed: %s", exc) return {"total": 0, "critical": 0, "high": 0, "avg_disruption": 0.0, "by_dim": {}, "status": "error"} @_flask_cache.memoize(timeout=30) def _load_graph_elements_cached() -> list[dict]: """Cached graph.json parse — prevents re-reading file on every graph tab load.""" return _load_graph_elements() _TABS = [ ("overview", "Field Intelligence"), ("radar", "Disruption Horizon"), ("feed", "Signal Feed"), ("chatbot", "Strategic Advisor"), ("graph", "Knowledge Graph"), ("reports", "Strategic Reports"), ("lens", "Intelligence Lens"), ] # ── Sidebar ──────────────────────────────────────────────────── sidebar = html.Div([ html.Div([ html.Div("SENTINEL", className="sb-brand-name"), html.Div("AGRO-MARKET INTELLIGENCE", className="sb-brand-sub"), ], className="sb-brand"), dcc.Loading( html.Div(id="sidebar-body"), id="sidebar-body-loading", type="dot", color="#7d8fa8", style={"minHeight": "40px"}, ), # live metrics html.Div([ dbc.Button("Run Scout Now", id="run-scout-btn", className="btn-scout", color="success", size="sm", outline=True), html.Div(id="run-scout-status", className="sb-run-status"), ], className="sb-action"), html.Div([ html.Div("v4.0 · EU Data Act 2026"), html.Div("AGCO / Fendt Intelligence Platform"), ], className="sb-footer"), ], className="war-sidebar") # ── Top Bar ──────────────────────────────────────────────────── topbar = html.Header([ html.Div("Fendt PESTEL-EL Strategic Sentinel", className="topbar-title"), html.Div(id="topbar-badge"), dbc.Button("Export Report", id="export-btn", className="btn-refresh", color="secondary", size="sm", outline=True, style={"marginRight": "8px"}), dbc.Button("Refresh", id="refresh-btn", className="btn-refresh", color="secondary", size="sm", outline=True), html.Div(id="topbar-ts", className="topbar-ts"), ], className="war-topbar") # ── Full Layout ──────────────────────────────────────────────── def _layout() -> html.Div: return html.Div([ sidebar, html.Div([ topbar, html.Nav( dbc.Tabs( [dbc.Tab(label=lbl, tab_id=tid) for tid, lbl in _TABS], id="main-tabs", active_tab="overview", className="war-tabs", ), className="war-tabnav", ), dcc.Loading( html.Div(id="page-canvas", className="war-canvas"), id="page-canvas-loading", type="circle", color="#00e5ff", style={"position": "relative"}, ), ], className="war-main"), # Persistent state dcc.Store(id="chat-store", data=[]), dcc.Store(id="signals-store", data=[], storage_type="memory"), dcc.Store(id="reports-last-selection", data=None), dcc.Store(id="chip-echo-store", data=""), dcc.Download(id="export-download"), dcc.Download(id="reports-pdf-download"), # 30-second auto-refresh (sponsor requirement #5) dcc.Interval(id="interval-30s", interval=6 * 60 * 60 * 1_000, n_intervals=0), # Force dash-cytoscape JS bundle to load on initial page render. # Without this, the bundle is absent when the graph tab is first clicked. html.Div( cyto.Cytoscape(id="dummy-cyto", elements=[], layout={"name": "preset"}), style={"display": "none"}, ), ], className="war-shell") app.layout = _layout() # ── Start scheduler at module load (works under gunicorn and __main__) ── # Must be AFTER app.layout so Dash is fully initialised. # gunicorn imports this module directly — __main__ guard would skip it. _scheduler_engine.start() atexit.register(_scheduler_engine.stop) log.info("SchedulerEngine started at module load. Next scout: %s", HEALTH.get("next_run_utc")) # ───────────────────────────────────────────────────────────── # Callbacks # ───────────────────────────────────────────────────────────── @app.callback( Output("signals-store", "data"), Input("interval-30s", "n_intervals"), Input("refresh-btn", "n_clicks"), ) def refresh_signals_store(_i: int, _n: int) -> list[dict]: """Populate signals-store with top-50 signals every 30 s for cross-tab access.""" try: db = _get_db() total = db.count() if total == 0: return [] results = db.search("EU agricultural market", n_results=min(50, total)) return [ { "id": s.id, "title": s.title, "dim": s.pestel_dimension.value, "score": s.disruption_score, } for s, _ in results ] except Exception as exc: log.error("refresh_signals_store failed: %s", exc) return [] @app.callback( Output("page-canvas", "children"), Input("main-tabs", "active_tab"), Input("interval-30s", "n_intervals"), Input("refresh-btn", "n_clicks"), State("chat-store", "data"), ) def render_tab(tab: str, _i: int, _n: int, history: list) -> html.Div: triggered = callback_context.triggered_id # Chatbot and Knowledge Graph only re-render on explicit tab switch. # Chatbot: avoids clobbering live conversation on interval ticks. # Graph: avoids rerunning the expensive Cytoscape COSE layout every 30s. if tab == "chatbot": return no_update if triggered != "main-tabs" else _tab_chatbot(history or []) if tab == "graph": if triggered == "interval-30s": return no_update try: # Surface active rebuild status (or last result) when switching to/refreshing the tab. return _tab_graph(status=_GRAPH_REBUILD.get("status", "")) except Exception as exc: log.error("render_tab(graph) crashed: %s", exc, exc_info=True) return html.Div( f"Render error in 'graph' — check logs for details: {exc}", style={"color": "#ff6090", "padding": "24px", "fontFamily": "JetBrains Mono, monospace", "fontSize": "12px"}, ) dispatch = { "overview": _tab_overview, "radar": _tab_radar, "feed": _tab_feed, "graph": _tab_graph, "reports": _tab_reports, "lens": _tab_lens, } try: return dispatch.get(tab, _tab_overview)() except Exception as exc: log.error("render_tab(%s) crashed: %s", tab, exc, exc_info=True) return html.Div( f"Render error in '{tab}' — check logs for details: {exc}", style={"color": "#ff6090", "padding": "24px", "fontFamily": "JetBrains Mono, monospace", "fontSize": "12px"}, ) @app.callback( Output("radar-chart", "figure"), Output("radar-table-container", "children"), Input("radar-dim-filter", "value"), Input("radar-score-slider", "value"), Input("interval-30s", "n_intervals"), Input("refresh-btn", "n_clicks"), ) def update_radar(dim_filter: str, min_score: float, _i: int, _n: int): try: signals = _get_unique_signals_cached() fig = _chart_radar(signals, dim_filter or "All", min_score or 0.50) filtered = [s for s in signals if (dim_filter in (None, "All") or s.pestel_dimension.value == dim_filter) and s.disruption_score >= (min_score or 0.50)] filtered.sort(key=lambda s: s.disruption_score, reverse=True) if not filtered: table = html.Div("No signals match current filters.", style={"fontSize": "11px", "color": "#6a8099", "padding": "12px 0"}) else: table = html.Table([ html.Thead(html.Tr([ html.Th("Dim"), html.Th("Signal Title"), html.Th("Score"), html.Th("Src"), ])), html.Tbody([ html.Tr([ html.Td(html.Span( _DIM_PILL_CODE.get(s.pestel_dimension.value, "?"), className=f"dim-pill dp-{_DIM_PILL_CODE.get(s.pestel_dimension.value, 'P')}", )), html.Td(s.title, style={"color": "#e8edf5", "fontSize": "12px"}), html.Td(f"{s.disruption_score:.3f}", style={"fontFamily": "JetBrains Mono, monospace", "fontSize": "11px", "color": _SEV_COLOUR.get(_sev(s.disruption_score))}), html.Td(html.A("↗ Source", href=s.source_url, target="_blank", className="source-link")), ]) for s in filtered[:50] ]), ], className="war-table") return fig, table except Exception as exc: log.error("update_radar crashed: %s", exc, exc_info=True) return go.Figure(), html.Div("Error loading radar table.") @app.callback( Output("feed-table-body", "children"), Output("feed-count-label", "children"), Input("feed-sort-dropdown", "value"), Input("feed-dim-dropdown", "value"), Input("interval-30s", "n_intervals"), Input("refresh-btn", "n_clicks"), ) def update_feed(sort_by: str, dim_filter: str, _i: int, _n: int): try: signals = _get_unique_signals_cached() if dim_filter and dim_filter != "ALL": signals = [s for s in signals if s.pestel_dimension.value == dim_filter] if sort_by == "score_desc": signals = sorted(signals, key=lambda s: s.disruption_score, reverse=True) elif sort_by == "score_asc": signals = sorted(signals, key=lambda s: s.disruption_score) else: signals = sorted(signals, key=lambda s: s.date_ingested, reverse=True) label = f"{len(signals)} signal(s) · {(sort_by or 'newest').replace('_', ' ')} · live from Astra DB" return [_row(s) for s in signals[:100]], label except Exception as exc: log.error("update_feed crashed: %s", exc, exc_info=True) return [], f"Error loading signals: {exc}" @app.callback( Output("sidebar-body", "children"), Output("topbar-badge", "children"), Output("topbar-ts", "children"), Input("interval-30s", "n_intervals"), Input("refresh-btn", "n_clicks"), ) def update_sidebar(_i: int, _n: int): stats = _db_stats_cached() total = stats["total"] by_dim = stats.get("by_dim", {}) db_kind = "live" if total else "idle" gem_kind = "live" if _HF_OK else "warn" sched_kind = "live" if HEALTH["scheduler_alive"] else "idle" scout_kind = "warn" if HEALTH["scout_running"] else sched_kind body = html.Div([ html.Div([ html.Div("ANALYTICS", className="sb-section-label"), html.Div([ html.Div("Signals", className="sb-kpi-label"), html.Div(str(total) if total else "—", className="sb-kpi-value"), ], className="sb-kpi"), html.Div([ html.Div("Critical", className="sb-kpi-label"), html.Div(str(stats["critical"]) if total else "—", className="sb-kpi-value"), ], className="sb-kpi"), html.Div([ html.Div("Avg Score", className="sb-kpi-label"), html.Div(f'{stats["avg_disruption"]:.3f}' if total else "—", className="sb-kpi-value"), ], className="sb-kpi"), ], className="sb-section"), html.Div(className="sb-divider"), html.Div([ html.Div("SERVICES", className="sb-section-label"), _dot("Astra DB", db_kind), _dot("HuggingFace API", gem_kind), _dot("Scheduler", sched_kind), _dot("Scout", scout_kind), ], className="sb-section"), html.Div(className="sb-divider"), html.Div([ html.Div("COVERAGE", className="sb-section-label"), *[html.Div([ html.Span(d[:3], className="sb-cov-dim", style={"color": _DIM_COLOUR.get(d, "#7d8fa8")}), html.Span(str(by_dim.get(d, 0)), className="sb-cov-count"), ], className="sb-cov-row") for d in ["POLITICAL", "ECONOMIC", "SOCIAL", "TECHNOLOGICAL", "ENVIRONMENTAL", "LEGAL"]], ], className="sb-section"), ]) badge = html.Div( f"{total} signals" if total else "NO DATA", className="topbar-badge", style={ "color": "#00e676" if total else "#ff1744", "borderColor": "rgba(0,230,118,0.5)" if total else "rgba(255,23,68,0.5)", "background": "rgba(0,230,118,0.08)" if total else "rgba(255,23,68,0.08)", }, ) ts = datetime.now(timezone.utc).strftime("UTC %H:%M:%S · auto-refresh 6h") return body, badge, ts _CHIP_TEXTS = [ "Which signals should Fendt's sales team lead with in dealer conversations this quarter?", "How should Fendt marketing position the Vario tractor line against CNH and Deere given current EU signals?", "What precision farming trends give AGCO the strongest upsell narrative to existing customers?", "Which regulatory changes create urgency for farmers to upgrade equipment — and how do we message that?", "What competitive threats from John Deere, CNH, or Claas should Fendt sales reps be prepared to counter?", ] @app.callback( Output("chip-echo-store", "data"), [Input(f"chip-{i}", "n_clicks") for i in range(5)], prevent_initial_call=True, ) def _fill_input_from_chip(*_clicks): ctx = callback_context if not ctx.triggered: return no_update tid = ctx.triggered_id if tid and str(tid).startswith("chip-"): return _CHIP_TEXTS[int(str(tid).split("-")[1])] return no_update @app.callback( Output("chat-input", "value", allow_duplicate=True), Input("chip-echo-store", "data"), prevent_initial_call=True, ) def _echo_chip_to_input(text: str): return text if text else no_update @app.callback( output=[ Output("chat-messages", "children"), Output("chat-store", "data"), Output("chat-input", "value"), ], inputs=[ Input("chat-send", "n_clicks"), Input("chat-input", "n_submit"), Input("chip-0", "n_clicks"), Input("chip-1", "n_clicks"), Input("chip-2", "n_clicks"), Input("chip-3", "n_clicks"), Input("chip-4", "n_clicks"), ], state=[ State("chat-input", "value"), State("chat-store", "data"), ], running=[ (Output("chat-send", "disabled"), True, False), (Output("chat-input", "disabled"), True, False), ], prevent_initial_call=True, background=True, ) def send_message(n_send, n_sub, c0, c1, c2, c3, c4, question_val, history_data): chip_texts = _CHIP_TEXTS question = question_val or "" history = list(history_data or []) triggered = callback_context.triggered_id if triggered and str(triggered).startswith("chip-"): question = chip_texts[int(str(triggered).split("-")[1])] if not question.strip(): return no_update, no_update, no_update question = question.strip() try: results = _get_db().search(question, n_results=6) context = [sig for sig, _ in results] except Exception: context = [] # ── Multi-Agent routing ─────────────────────────────────────────────────── agent_result = run_agent_query(question, context) answer = agent_result.get("final_answer", "Agent returned no answer.") route = agent_result.get("route", "synthesis") trace = agent_result.get("agent_trace", []) confidence = agent_result.get("confidence", "medium") # Prepend route badge so the user can see which agent responded route_label = "QUANTITATIVE · Calculator" if route == "quantitative" else "SYNTHESIS · Analyst" conf_colour = {"high": "#00e676", "medium": "#ffd93d", "low": "#ff6090"}.get(confidence, "#7d8fa8") badge_text = f"[{route_label} · confidence={confidence} · agents={' → '.join(trace)}]" history.append({"role": "user", "text": question}) history.append({"role": "assistant", "text": answer, "badge": badge_text, "badge_colour": conf_colour}) if len(history) > 20: history = history[-20:] welcome = _chat_bubble( f"Fendt Relational Brain — Multi-Agent Strategic Advisor\n\n" f"{_db_stats_cached()['total']} signal(s) in Astra DB. " f"Router automatically directs queries to the Calculator Agent " f"(quantitative) or Analyst Agent (synthesis).", role="assistant", ) bubbles = [welcome] for msg in history: bubble = _chat_bubble(msg["text"], msg["role"]) if msg["role"] == "assistant" and msg.get("badge"): badge = html.Div( msg["badge"], style={ "fontSize": "9px", "fontFamily": "JetBrains Mono, monospace", "color": msg.get("badge_colour", "#7d8fa8"), "marginTop": "6px", "opacity": "0.75", }, ) bubble = html.Div([bubble, badge]) bubbles.append(bubble) return bubbles, history, "" @app.callback( Output("run-scout-status", "children"), Input("run-scout-btn", "n_clicks"), prevent_initial_call=True, ) def trigger_scout(n: int) -> str: if not _HF_OK: return "HuggingFace API token missing." _scheduler_engine.trigger_now() log.info("Manual scout triggered via UI (n_clicks=%d)", n) return "Scout running in background — check sidebar for updates." @app.callback( Output("page-canvas", "children", allow_duplicate=True), Input("rebuild-graph-btn", "n_clicks"), Input("run-inference-btn", "n_clicks"), prevent_initial_call=True, ) def graph_action(rebuild_n: int, infer_n: int): """Handle Rebuild Graph and Run Inference buttons.""" triggered = callback_context.triggered_id if triggered == "rebuild-graph-btn": if _GRAPH_REBUILD["running"]: try: return _tab_graph(status="Rebuild already in progress — click Refresh when complete.") except Exception as exc: log.error("graph_action _tab_graph failed: %s", exc) return html.Div(f"Graph render error: {exc}", style={"color": "#ff6090", "padding": "24px", "fontFamily": "JetBrains Mono, monospace"}) def _do_rebuild() -> None: _GRAPH_REBUILD["running"] = True _GRAPH_REBUILD["status"] = "Rebuilding in background…" try: counts = rebuild_graph_from_db() _flask_cache.delete_memoized(_load_graph_elements_cached) _GRAPH_REBUILD["status"] = ( f"Graph rebuilt: {counts['nodes']} nodes, " f"{counts['links']} edges, {counts['triples']} triples" ) log.info("graph_action: background rebuild complete — %s", _GRAPH_REBUILD["status"]) except Exception as exc: log.error("graph_action background rebuild failed: %s", exc) _GRAPH_REBUILD["status"] = f"Rebuild failed: {exc}" finally: _GRAPH_REBUILD["running"] = False _threading.Thread(target=_do_rebuild, daemon=True).start() try: return _tab_graph(status="Rebuilding in background — click Refresh when complete (~60 s).") except Exception as exc: log.error("graph_action _tab_graph failed: %s", exc) return html.Div(f"Graph render error: {exc}", style={"color": "#ff6090", "padding": "24px", "fontFamily": "JetBrains Mono, monospace"}) elif triggered == "run-inference-btn": try: result = infer_hidden_relationships() _flask_cache.delete_memoized(_load_graph_elements_cached) added = result["inferred_added"] total = result["total_triples"] status = f"Inference complete: +{added} hidden cascades ({total} total triples)" except Exception as exc: log.error("graph_action inference failed: %s", exc) status = f"Inference failed: {exc}" try: return _tab_graph(status=status) except Exception as exc: log.error("graph_action _tab_graph failed: %s", exc) return html.Div(f"Graph render error: {exc}", style={"color": "#ff6090", "padding": "24px", "fontFamily": "JetBrains Mono, monospace"}) return no_update @app.callback( Output("export-download", "data"), Input("export-btn", "n_clicks"), prevent_initial_call=True, ) def export_report(n_clicks: int): try: html_content = _build_export_html() filename = f"fendt-pestel-report-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M')}.html" return dcc.send_string(html_content, filename) except Exception as exc: log.error("export_report failed: %s", exc) return no_update # ── Strategic Reports callback ───────────────────────── @app.callback( Output("reports-body", "children"), Input("reports-dropdown", "value"), prevent_initial_call=True, ) def render_report(path: str | None) -> html.Div: if not path: return no_update # Only render if it's a real user selection, not the initial default # (prevents duplicate rendering on tab load) return _render_report_body(path) @app.callback( Output("reports-pdf-download", "data"), Output("reports-last-selection", "data"), Input("reports-export-pdf-btn", "n_clicks"), Input("reports-dropdown", "value"), State("reports-last-selection", "data"), prevent_initial_call=True, ) def export_report_pdf(n_clicks: int, current_path: str | None, last_path: str | None): triggered = callback_context.triggered_id # If the dropdown changed, update the last_selection and DO NOT download if triggered == "reports-dropdown": return no_update, current_path # If the button was clicked, verify it's a real click and then download if triggered == "reports-export-pdf-btn" and n_clicks: if not current_path or not _PDF_OK: return no_update, no_update try: content = Path(current_path).read_text(encoding="utf-8") pdf_bytes = _md_to_pdf_bytes(content) filename = f"{Path(current_path).stem}-{datetime.now(timezone.utc).strftime('%Y%m%d')}.pdf" return dcc.send_bytes(pdf_bytes, filename), current_path except Exception as exc: log.error("export_report_pdf failed: %s", exc) return no_update, current_path return no_update, current_path # ── Generate Intelligence Brief callback (background) ───────── @app.callback( output=[ Output("reports-dropdown", "options"), Output("reports-dropdown", "value"), Output("reports-gen-status", "children"), ], inputs=[Input("reports-gen-btn", "n_clicks")], running=[ (Output("reports-gen-btn", "disabled"), True, False), ( Output("reports-gen-status", "children"), html.Span("⚙ Generating brief — LLM working…", style={"color": "#ffd93d", "fontSize": "11px"}), "", ), ], prevent_initial_call=True, background=True, ) def generate_intelligence_brief(n_clicks: int): """Fetch top 10 signals, call generate_brief_markdown, write .md, refresh dropdown. Setting reports-dropdown.value triggers render_report automatically — no need to also output reports-body.children (that would be a duplicate output). """ try: db = _get_db() total = db.count() if total == 0: return no_update, no_update, "No signals in database — run Scout first." results = db.search("agricultural market disruption EU Fendt", n_results=min(10, total)) signals = sorted([sig for sig, _ in results], key=lambda s: s.disruption_score, reverse=True) md_text = generate_brief_markdown(signals) _REPORTS_DIR.mkdir(parents=True, exist_ok=True) ts_str = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") out_path = _REPORTS_DIR / f"Strategic_Brief_{ts_str}.md" out_path.write_text(md_text, encoding="utf-8") log.info("Generated brief: %s", out_path.name) new_options = _glob_reports() new_value = str(out_path) status_msg = f"✓ Brief generated: {out_path.name}" return new_options, new_value, status_msg except Exception as exc: log.error("generate_intelligence_brief failed: %s", exc) return no_update, no_update, f"Error: {exc}" # ── Intelligence Lens callback ──────────────────────────────── @app.callback( Output("lens-results", "children"), Input("lens-topic-dropdown", "value"), Input("lens-custom-input", "value"), Input("lens-custom-input", "n_submit"), prevent_initial_call=True, # initial content embedded by _tab_lens() ) def lens_search(topic: str | None, custom: str | None, _ns: int) -> html.Div: return _run_lens_search(topic, custom) # ───────────────────────────────────────────────────────────── # Entry point # ───────────────────────────────────────────────────────────── if __name__ == "__main__": _preflight() # Scheduler already started at module load — just log status stats = _db_stats_cached() log.info("App starting — Astra DB: %d signals, HuggingFace: %s", stats["total"], "OK" if _HF_OK else "NO KEY") print(f"\n Fendt Sentinel · http://localhost:8050") print(f" Astra DB : {stats['total']} signal(s)") print(f" HuggingFace: {'OK' if _HF_OK else 'no API key — set HUGGINGFACEHUB_API_TOKEN'}") print(f" Scheduler: active (6-hour scout cycle)") print(f" Auto-refresh: 30 seconds\n") # Use PORT env var (Hugging Face / Render) or default to 7860 port = int(os.environ.get("PORT", 7860)) app.run(debug=False, host="0.0.0.0", port=port)