app.py

from dash import Dash, html, dcc, Input, Output, State, no_update, ctx
import pandas as pd
import dash_mantine_components as dmc
import time

from config import DATASET_ID
from data_utils import create_fresh_duckdb_with_views, get_last_updated
from graphs.leaderboard import (
    leaderboard_callback_logic,
)
from helpers import build_slider_marks, get_thumb_label_single, get_thumb_labels
from layout_components import (
    build_main_layout,
    build_range_slider,
    build_single_slider,
    DATA_URL,
)

# Initialize the app
app = Dash(suppress_callback_exceptions=True) # suppress callback exceptions for multi-page layout
server = app.server

# Load dataset and determine time range
print(f"Attempting to connect to dataset from Hugging Face Hub: {DATASET_ID}")
try:
    overall_start_time = time.time()

    # Create fresh connection, views, and read start/end time
    conn = create_fresh_duckdb_with_views()
    try:
        time_range = conn.execute("SELECT MIN(time) AS min_time, MAX(time) AS max_time FROM all_downloads;").fetchdf()
        start_dt = pd.to_datetime(time_range["min_time"].iloc[0])
        end_dt = pd.to_datetime(time_range["max_time"].iloc[0])
    finally:
        conn.close()

    msg = f"Successfully connected to datasets in {time.time() - overall_start_time:.2f}s."
    print(msg)
except Exception as e:
    err_msg = f"Failed to load dataset(s). Error: {e}"
    print(err_msg)
    raise

# Create a dcc slider for time range selection by year (readable marks)
start_ts = int(start_dt.timestamp())
end_ts = int(end_dt.timestamp())
marks = build_slider_marks(start_dt, end_dt)

time_slider = build_range_slider(
    start_ts,
    end_ts,
    [start_ts, end_ts],
    marks,
    thumb_children=get_thumb_labels([start_ts, end_ts]),
)

time_slider_alltime = build_single_slider(
    start_ts,
    end_ts,
    end_ts,
    marks,
    thumb_children=get_thumb_label_single(end_ts),
)

last_updated_label = get_last_updated()

# Define the app layout
app.layout = dmc.MantineProvider(
    theme={
        "colorScheme": "light",
        "primaryColor": "blue",
        "fontFamily": "Inter, sans-serif",
    },
    children=[
        html.A(id="paper-redirect", style={"display": "none"}),
        html.Div(id="data-link-open-noop", style={"display": "none"}),
        dcc.Store(id="selected-view", data="all_downloads"),
        dcc.Store(id="model-attribution-type", data="uploader"),
        dcc.Store(id="time-slider-value", data=[start_ts, end_ts]),
        dcc.Store(id="time-slider-alltime-value", data=end_ts),
        build_main_layout(last_updated_label, time_slider),
    ],
)

# ----------
# Callbacks
# ----------

# Update model attribution type based on user selection
@app.callback(
    Output("model-attribution-type", "data"),
    Input("model-attribution-segmented", "value"),
)
def update_model_attribution_type(selected_value):
    return selected_value

# Toggle between range slider and all-time slider
@app.callback(
    Output("slider-container", "children"),
    Output("slider-description", "children"),
    Input("time-range-toggle", "checked"),
    State("time-slider-value", "data"),
    State("time-slider-alltime-value", "data"),
)
def toggle_time_slider(is_alltime, range_value, alltime_value):
    if is_alltime:
        description = "Select a specific date to view all-time cumulative downloads up to that point."
        slider_value = alltime_value if alltime_value is not None else end_ts
        slider = build_single_slider(
            start_ts,
            end_ts,
            slider_value,
            marks,
            thumb_children=get_thumb_label_single(slider_value),
        )
        return [slider], description
    else:
        description = "Adjust the slider to filter leaderboard results by the difference in downloads within the time range."
        slider_val = range_value if range_value is not None else [start_ts, end_ts]
        slider = build_range_slider(
            start_ts,
            end_ts,
            slider_val,
            marks,
            thumb_children=get_thumb_labels(slider_val),
        )
        return [slider], description

# Sync slider values to dcc.Store components
@app.callback(
    Output("time-slider-value", "data"),
    Input("time-slider", "value"),
)
def sync_time_slider_value(value):
    return value

# Sync all-time slider value to dcc.Store component
@app.callback(
    Output("time-slider-alltime-value", "data"),
    Input("time-slider-alltime", "value"),
)
def sync_time_slider_alltime_value(value):
    return value

# Update Top Countries leaderboard
@app.callback(
    Output("top_countries-table", "children"),
    Output("top_countries-toggle", "children"),
    Input("top_countries-toggle", "n_clicks"),
    Input("time-slider-value", "data"),
    Input("time-slider-alltime-value", "data"),
    Input("time-range-toggle", "checked"),
    Input("selected-view", "data"),
    Input("model-attribution-type", "data"),
    State("top_countries-toggle", "children"),
)
def update_top_countries(
    n_clicks, slider_value, slider_alltime_value, is_alltime, selected_view, attribution_type, current_label
):
    # Use the appropriate slider value based on the toggle
    active_slider_value = slider_alltime_value if is_alltime else slider_value
    
    return leaderboard_callback_logic(
        n_clicks,
        active_slider_value,
        current_label,
        group_col="org_country_single",
        filename="top_countries",
        default_label="▼ Show Top 50",
        chip_color="#F0F9FF",
        view=selected_view,
        derived_author_toggle=(attribution_type == "original_creator"),
        is_alltime=is_alltime,
    )

# Update Top Developers leaderboard
@app.callback(
    Output("top_developers-table", "children"),
    Output("top_developers-toggle", "children"),
    Input("top_developers-toggle", "n_clicks"),
    Input("time-slider-value", "data"),
    Input("time-slider-alltime-value", "data"),
    Input("time-range-toggle", "checked"),
    Input("selected-view", "data"),
    Input("model-attribution-type", "data"),
    State("top_developers-toggle", "children"),
)
def update_top_developers(
    n_clicks, slider_value, slider_alltime_value, is_alltime, selected_view, attribution_type, current_label
):
    # Use derived_author if attribution_type == "original_creator", else author
    group_col = "derived_author" if attribution_type == "original_creator" else "author"
    # Use the appropriate slider value based on the toggle
    active_slider_value = slider_alltime_value if is_alltime else slider_value
    
    return leaderboard_callback_logic(
        n_clicks,
        active_slider_value,
        current_label,
        group_col=group_col,
        filename="top_developers",
        default_label="▼ Show Top 50",
        chip_color="#F0F9FF",
        view=selected_view,
        derived_author_toggle=(attribution_type == "original_creator"),
        is_alltime=is_alltime,
    )

# Update Top Models leaderboard
@app.callback(
    Output("top_models-table", "children"),
    Output("top_models-toggle", "children"),
    Input("top_models-toggle", "n_clicks"),
    Input("time-slider-value", "data"),
    Input("time-slider-alltime-value", "data"),
    Input("time-range-toggle", "checked"),
    Input("selected-view", "data"),
    Input("model-attribution-type", "data"),
    State("top_models-toggle", "children"),
)
def update_top_models(
    n_clicks, slider_value, slider_alltime_value, is_alltime, selected_view, attribution_type, current_label
):
    # Use the appropriate slider value based on the toggle
    active_slider_value = slider_alltime_value if is_alltime else slider_value
    
    return leaderboard_callback_logic(
        n_clicks,
        active_slider_value,
        current_label,
        group_col="model",
        filename="top_models",
        default_label="▼ Show More",
        chip_color="#F0F9FF",
        view=selected_view,
        derived_author_toggle=(attribution_type == "original_creator"),
        is_alltime=is_alltime,
    )

# Update thumb labels for range slider
@app.callback(
    Output("time-slider", "thumbChildren"),
    Input("time-slider", "value"),
)
def update_thumb_labels(values):
    return get_thumb_labels(values)

# Update thumb label for all-time slider
@app.callback(
    Output("time-slider-alltime", "thumbChildren"),
    Input("time-slider-alltime", "value"),
)
def update_thumb_label_alltime(value):
    return get_thumb_label_single(value)

# Update selected view based on segmented control
@app.callback(
    Output("selected-view", "data"),
    Input("segmented", "value"),
)
def update_selected_view(seg_value):
    if seg_value == "filtered-downloads":
        return "one_year_rolling"
    return "all_downloads"


@app.callback(
    Output("paper-disclaimer-modal", "opened"),
    Output("paper-redirect", "href"),
    Input("data-link-trigger-header", "n_clicks"),
    Input("paper-disclaimer-close", "n_clicks"),
    Input("paper-disclaimer-confirm", "n_clicks"),
    State("paper-disclaimer-modal", "opened"),
    prevent_initial_call=True,
)
def handle_paper_disclaimer(
    data_clicks,
    close_clicks,
    confirm_clicks,
    modal_opened,
):
    triggered_id = ctx.triggered_id

    if triggered_id == "data-link-trigger-header":
        return True, no_update

    if triggered_id == "paper-disclaimer-close":
        return False, no_update

    if triggered_id == "paper-disclaimer-confirm":
        return False, DATA_URL

    return modal_opened, no_update


app.clientside_callback(
    """
    function(url) {
        if (url) {
            window.open(url, "_blank", "noopener,noreferrer");
        }
        return "";
    }
    """,
    Output("data-link-open-noop", "children"),
    Input("paper-redirect", "href"),
    prevent_initial_call=True,
)

# Run the app
if __name__ == "__main__":
    app.run(debug=True)