"""
app.py
HIFFI Artist Intelligence - Executive Dashboard
Clean dark-themed design with aligned dataframes, responsive cards, and scouting dossier.
"""
import io
import os
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
import streamlit as st
st.set_page_config(
page_title="HIFFI | Artist Intelligence",
page_icon="⚡",
layout="wide",
initial_sidebar_state="expanded",
)
DATA_PATH = "enriched_artists.csv"
# Global CSS Polish
st.markdown(
"""
""",
unsafe_allow_html=True,
)
@st.cache_data(ttl=60)
def load_data():
if not os.path.exists(DATA_PATH):
return pd.DataFrame()
df = pd.read_csv(DATA_PATH)
numeric_cols = [
"hiffi_score",
"breakout_velocity_index",
"subscriber_count",
"daily_view_velocity",
"momentum_30d",
"confidence_score",
"hiffi_rank",
"breakout_rank",
]
for col in numeric_cols:
if col in df.columns:
df[col] = pd.to_numeric(df[col], errors="coerce")
# Fallback assignment for subgenre if not yet computed in pipeline
if "subgenre" not in df.columns:
def infer_subgenre(r):
subs = r.get("subscriber_count", 0) or 0
bvi = r.get("breakout_velocity_index", 0) or 0
if bvi >= 70:
return "Breakout / Emerging"
elif subs >= 15000000:
return "Legends"
else:
return "Trap & Street"
df["subgenre"] = df.apply(infer_subgenre, axis=1)
return df
def to_excel_bytes(dataframe: pd.DataFrame) -> bytes:
"""Converts dataframe to downloadable in-memory Excel spreadsheet."""
output = io.BytesIO()
with pd.ExcelWriter(output, engine="openpyxl") as writer:
dataframe.to_excel(writer, index=False, sheet_name="HIFFI_Cohort")
return output.getvalue()
df = load_data()
# ----------------- Sidebar Controls -----------------
with st.sidebar:
st.markdown("### ⚡ HIFFI Intelligence")
st.caption("Hip-Hop Market Scoring & Breakout Detection")
st.markdown("
", unsafe_allow_html=True)
if df.empty:
st.error(f"'{DATA_PATH}' not found. Please run 'python main.py' first!")
st.stop()
view_mode = st.radio(
"Leaderboard Perspective:",
["🏆 Overall HIFFI Rank", "🚀 Breakout Velocity (BVI)", "🔍 Master Audit Grid"],
index=0,
)
st.markdown("", unsafe_allow_html=True)
st.markdown("##### 🛡️ Filters")
# Subgenre / Vertical Filter
all_verticals = sorted(df["subgenre"].dropna().unique().tolist())
target_presets = ["Legends", "Trap & Street", "UK Drill", "Lyricists", "Breakout / Emerging"]
vertical_choices = [v for v in target_presets if v in all_verticals] + [v for v in all_verticals if v not in target_presets]
selected_verticals = st.multiselect(
"Hip-Hop Vertical",
options=vertical_choices,
default=vertical_choices,
)
available_statuses = df["health_status"].dropna().unique().tolist()
selected_statuses = st.multiselect(
"Channel Status",
options=available_statuses,
default=available_statuses,
)
min_conf = st.slider(
"Min Confidence",
min_value=0.0,
max_value=1.0,
value=0.50,
step=0.05,
)
# Filter logic
filtered_df = df[
(df["health_status"].isin(selected_statuses))
& (df["subgenre"].isin(selected_verticals))
& (df["confidence_score"] >= min_conf)
].copy()
# Sidebar Data Export Buttons (Filters applied)
with st.sidebar:
st.markdown("", unsafe_allow_html=True)
st.markdown("##### 💾 Export Filtered Cohort")
col_csv, col_xlsx = st.columns(2)
csv_payload = filtered_df.to_csv(index=False).encode("utf-8")
col_csv.download_button(
label="📥 CSV",
data=csv_payload,
file_name="hiffi_filtered_cohort.csv",
mime="text/csv",
use_container_width=True,
)
excel_payload = to_excel_bytes(filtered_df)
col_xlsx.download_button(
label="📊 Excel",
data=excel_payload,
file_name="hiffi_filtered_cohort.xlsx",
mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
use_container_width=True,
)
st.markdown("", unsafe_allow_html=True)
st.caption("Momentum (35%) • Velocity (30%) • Reach (20%) • Cadence (15%)")
# ----------------- Main Header -----------------
head_col1, head_col2 = st.columns([3, 1])
with head_col1:
st.markdown("HIFFI Artist Intelligence Hub
", unsafe_allow_html=True)
st.markdown("Cross-platform tracking of catalog sustainability and audience breakout velocity.
", unsafe_allow_html=True)
with head_col2:
st.markdown(
f"Active Cohort: {len(filtered_df)} / {len(df)} artists
",
unsafe_allow_html=True,
)
# ----------------- KPI Cards Row -----------------
k1, k2, k3, k4 = st.columns(4)
with k1:
st.markdown(
f"""
Tracked Artists
{len(df)}
Monitored pipeline size
""",
unsafe_allow_html=True,
)
with k2:
active_count = len(df[df["health_status"] == "ACTIVE_HEALTHY"])
st.markdown(
f"""
Active Healthy
{active_count}
Recent uploads & steady traction
""",
unsafe_allow_html=True,
)
with k3:
evergreen_count = len(df[df["health_status"] == "CATALOG_EVERGREEN"])
st.markdown(
f"""
Catalog Evergreen
{evergreen_count}
Strong passive catalog streams
""",
unsafe_allow_html=True,
)
with k4:
zombies = len(df[df["health_status"] == "DORMANT_CATALOG"])
st.markdown(
f"""
Dormant / Excluded
{zombies}
Disqualified (<NA> Rank)
""",
unsafe_allow_html=True,
)
st.markdown("", unsafe_allow_html=True)
# ----------------- Top 3 Leaders Podium -----------------
top3 = (
filtered_df[filtered_df["is_rank_eligible"] == True]
.sort_values(by="hiffi_rank", ascending=True)
.head(3)
)
if len(top3) >= 3:
st.markdown("#### 👑 Market Leaders")
p1, p2, p3 = st.columns(3)
podium_data = [
(p1, top3.iloc[0], "🥇 RANK #1", "podium-gold", "#f59e0b"),
(p2, top3.iloc[1], "🥈 RANK #2", "podium-silver", "#94a3b8"),
(p3, top3.iloc[2], "🥉 RANK #3", "podium-bronze", "#f97316"),
]
for col, row_data, rank_label, css_class, accent_color in podium_data:
badge_cls = (
"badge-active"
if row_data["health_status"] == "ACTIVE_HEALTHY"
else "badge-evergreen"
)
bvi_str = f"{row_data['breakout_velocity_index']:.1f}" if pd.notna(row_data.get("breakout_velocity_index")) else "N/A"
vel_str = f"{int(row_data['daily_view_velocity']):,}" if pd.notna(row_data.get("daily_view_velocity")) else "N/A"
with col:
st.markdown(
f"""
{rank_label}
{row_data['health_status'].replace('_', ' ')}
{row_data['name']}
{row_data.get('channel_title', '')}
HIFFI Score
{row_data['hiffi_score']:.1f}
""",
unsafe_allow_html=True,
)
st.markdown("", unsafe_allow_html=True)
# ----------------- Leaderboard Grid -----------------
st.markdown(f"#### 📋 {view_mode}")
if view_mode == "🏆 Overall HIFFI Rank":
table_df = (
filtered_df[filtered_df["is_rank_eligible"] == True]
.sort_values(by="hiffi_rank", ascending=True)
.copy()
)
cols = [
"hiffi_rank",
"name",
"subgenre",
"hiffi_score",
"breakout_rank",
"breakout_velocity_index",
"daily_view_velocity",
"momentum_30d",
"health_status",
]
elif view_mode == "🚀 Breakout Velocity (BVI)":
table_df = (
filtered_df[
(filtered_df["is_rank_eligible"] == True)
& (filtered_df["breakout_rank"].notna())
]
.sort_values(by="breakout_rank", ascending=True)
.copy()
)
cols = [
"breakout_rank",
"name",
"subgenre",
"breakout_velocity_index",
"hiffi_rank",
"hiffi_score",
"daily_view_velocity",
"subscriber_count",
"health_status",
]
else:
table_df = filtered_df.sort_values(by=["is_rank_eligible", "hiffi_score"], ascending=[False, False]).copy()
cols = [
"name",
"subgenre",
"hiffi_score",
"breakout_velocity_index",
"daily_view_velocity",
"subscriber_count",
"health_status",
"confidence_score",
"is_rank_eligible",
]
# Keep only columns that exist in DataFrame to prevent KeyError
render_cols = [c for c in cols if c in table_df.columns]
st.dataframe(
table_df[render_cols].reset_index(drop=True),
use_container_width=True,
height=290,
column_config={
"hiffi_rank": st.column_config.NumberColumn("Rank", format="%d", width="small"),
"breakout_rank": st.column_config.NumberColumn("BVI Rank", format="%d", width="small"),
"name": st.column_config.TextColumn("Artist", width="medium"),
"subgenre": st.column_config.TextColumn("Vertical", width="medium"),
"hiffi_score": st.column_config.ProgressColumn(
"HIFFI Score",
format="%.1f",
min_value=0,
max_value=100,
width="medium",
),
"breakout_velocity_index": st.column_config.ProgressColumn(
"Breakout (BVI)",
format="%.1f",
min_value=0,
max_value=100,
width="medium",
),
"daily_view_velocity": st.column_config.NumberColumn(
"Daily Views / Day",
format="%,d",
width="medium",
),
"momentum_30d": st.column_config.NumberColumn(
"30d Momentum",
format="%.1f%%",
width="small",
),
"subscriber_count": st.column_config.NumberColumn(
"Subscribers",
format="%,d",
width="medium",
),
"health_status": st.column_config.TextColumn("Status", width="medium"),
"confidence_score": st.column_config.NumberColumn("Confidence", format="%.2f", width="small"),
"is_rank_eligible": st.column_config.CheckboxColumn("Eligible", width="small"),
},
)
st.markdown("", unsafe_allow_html=True)
# ----------------- Visual Insights Charts -----------------
st.markdown("#### 📊 Market Landscape Visualizer")
c_left, c_right = st.columns(2)
with c_left:
scatter_df = filtered_df[filtered_df["is_rank_eligible"] == True].copy()
fig_scatter = px.scatter(
scatter_df,
x="daily_view_velocity",
y="momentum_30d",
size="subscriber_count",
color="health_status",
color_discrete_map={
"ACTIVE_HEALTHY": "#10b981",
"CATALOG_EVERGREEN": "#38bdf8",
"DORMANT_CATALOG": "#ef4444",
},
hover_name="name",
hover_data={"daily_view_velocity": ":,d", "momentum_30d": ":.1f%"},
log_x=True,
title="View Velocity vs. 30-Day Momentum",
labels={
"daily_view_velocity": "Daily Views / Day (Log Scale)",
"momentum_30d": "30-Day Growth (%)",
"health_status": "Vitality",
},
template="plotly_dark",
)
fig_scatter.update_layout(
paper_bgcolor="#111827",
plot_bgcolor="#111827",
font=dict(color="#9ca3af"),
margin=dict(l=20, r=20, t=40, b=20),
height=320,
)
st.plotly_chart(fig_scatter, use_container_width=True)
with c_right:
bvi_df = (
filtered_df[filtered_df["breakout_velocity_index"].notna()]
.sort_values(by="breakout_velocity_index", ascending=True)
)
fig_bar = px.bar(
bvi_df,
x="breakout_velocity_index",
y="name",
orientation="h",
color="breakout_velocity_index",
color_continuous_scale="Purples",
title="Breakout Velocity Index (BVI)",
labels={"breakout_velocity_index": "BVI Score", "name": "Artist"},
template="plotly_dark",
)
fig_bar.update_layout(
paper_bgcolor="#111827",
plot_bgcolor="#111827",
font=dict(color="#9ca3af"),
coloraxis_showscale=False,
margin=dict(l=20, r=20, t=40, b=20),
height=320,
)
st.plotly_chart(fig_bar, use_container_width=True)
st.markdown("", unsafe_allow_html=True)
# ----------------- Deep Dive Scouting Dossier -----------------
st.markdown("#### 🔎 Artist Dossier")
selected_name = st.selectbox("Inspect Artist Details:", df["name"].unique(), label_visibility="collapsed")
row = df[df["name"] == selected_name].iloc[0]
# Resolve YouTube channel direct link
yt_url = str(row.get("youtube_url", "")).strip()
yt_id = str(row.get("channel_id", row.get("youtube_channel_id", ""))).strip()
if (not yt_url or yt_url == "nan") and yt_id and yt_id != "nan":
if yt_id.startswith("UC"):
yt_url = f"https://www.youtube.com/channel/{yt_id}"
elif yt_id.startswith("@"):
yt_url = f"https://www.youtube.com/{yt_id}"
else:
yt_url = f"https://www.youtube.com/@{yt_id}"
d1, d2, d3, d4 = st.columns(4)
with d1:
h_val = row.get("hiffi_score")
st.metric("HIFFI Score", f"{h_val:.1f}" if pd.notna(h_val) else "N/A")
with d2:
b_val = row.get("breakout_velocity_index")
st.metric("Breakout Index (BVI)", f"{b_val:.1f}" if pd.notna(b_val) else "N/A")
with d3:
s_val = row.get("subscriber_count")
st.metric("Subscribers", f"{int(s_val):,}" if pd.notna(s_val) else "N/A")
with d4:
conf = row.get("confidence_score", 0.0)
tier = row.get("confidence_tier", "Standard")
st.metric("Entity Confidence", f"{conf * 100:.0f}%", delta=tier, delta_color="normal")
badge_cls = (
"badge-active"
if row.get("health_status") == "ACTIVE_HEALTHY"
else "badge-evergreen"
if row.get("health_status") == "CATALOG_EVERGREEN"
else "badge-dormant"
)
# Dossier info box with integrated Official YouTube Button
st.markdown(
f"""
Artist: {row['name']}
Vertical: {row.get('subgenre', 'N/A')}
Channel: {row.get('channel_title', 'Unknown')}
Status: {row.get('health_status', 'UNKNOWN')}
Zombie Flag: {'⚠️ Yes' if row.get('is_zombie') else '✅ Clean'}
Eligible for Rank: {'✅ Yes' if row.get('is_rank_eligible') else '❌ Disqualified'}
""",
unsafe_allow_html=True,
)
st.write("")
if yt_url and yt_url != "nan":
st.link_button("▶ Open Official YouTube Channel", yt_url, use_container_width=False)
else:
st.caption("⚠️ No verified YouTube channel link available for this artist.")