import gradio as gr
import pandas as pd
import plotly.express as px
from datetime import datetime, timedelta
import requests
from io import BytesIO
def create_trend_chart(space_id, daily_ranks_df):
if space_id is None or daily_ranks_df.empty:
return None
try:
space_data = daily_ranks_df[daily_ranks_df['id'] == space_id].copy()
if space_data.empty:
return None
space_data = space_data.sort_values('date')
fig = px.line(
space_data,
x='date',
y='rank',
title=f'Daily Rank Trend for {space_id}',
labels={'date': 'Date', 'rank': 'Rank'},
markers=True,
height=500 # 수정된 부분
)
fig.update_layout(
xaxis_title="Date",
yaxis_title="Rank",
yaxis=dict(
range=[100, 1],
tickmode='linear',
tick0=1,
dtick=10
),
hovermode='x unified',
plot_bgcolor='white',
paper_bgcolor='white',
showlegend=False,
margin=dict(t=50, r=20, b=40, l=40)
)
fig.update_xaxes(showgrid=True, gridwidth=1, gridcolor='lightgray')
fig.update_yaxes(showgrid=True, gridwidth=1, gridcolor='lightgray')
fig.update_traces(
line_color='#2563eb',
line_width=2,
marker=dict(size=8, color='#2563eb')
)
return fig
except Exception as e:
print(f"Error creating chart: {e}")
return None
def get_duplicate_spaces(top_100_spaces):
# ID에서 username/spacename 형식에서 username만 추출
top_100_spaces['clean_id'] = top_100_spaces['id'].apply(lambda x: x.split('/')[0])
# username별 trending score 합산
score_sums = top_100_spaces.groupby('clean_id')['trendingScore'].sum()
# 디버깅용 출력
print("\n=== ID별 스코어 합산 결과 ===")
for id, score in score_sums.sort_values(ascending=False).head(20).items():
print(f"ID: {id}, Total Score: {score}")
# 합산된 스코어로 정렬하여 상위 20개 선택
top_20_scores = score_sums.sort_values(ascending=False).head(20)
return top_20_scores
def create_duplicates_chart(score_sums):
if score_sums.empty:
return None
# 데이터프레임 생성
df = pd.DataFrame({
'id': score_sums.index,
'total_score': score_sums.values,
'rank': range(1, len(score_sums) + 1)
})
# 디버깅용 출력
print("\n=== 차트 데이터 ===")
print(df)
fig = px.bar(
df,
x='id',
y='rank',
title="Top 20 Spaces by Combined Trending Score",
height=500, # 수정된 부분
text='total_score'
)
fig.update_layout(
showlegend=False,
margin=dict(t=50, r=20, b=40, l=40),
plot_bgcolor='white',
paper_bgcolor='white',
xaxis_tickangle=-45,
yaxis=dict(
range=[20.5, 0.5],
tickmode='linear',
tick0=1,
dtick=1
)
)
fig.update_traces(
marker_color='#4CAF50',
texttemplate='%{text:.1f}',
textposition='outside',
hovertemplate='ID: %{x}
Rank: %{y}
Total Score: %{text:.1f}'
)
fig.update_xaxes(
title_text="User ID",
showgrid=True,
gridwidth=1,
gridcolor='lightgray'
)
fig.update_yaxes(
title_text="Rank",
showgrid=True,
gridwidth=1,
gridcolor='lightgray'
)
return fig
def update_display(selection):
global daily_ranks_df
if not selection:
return None, gr.HTML(value="Select a space to view details
")
try:
space_id = selection
latest_data = daily_ranks_df[
daily_ranks_df['id'] == space_id
].sort_values('date').iloc[-1]
info_text = f"""
Space Details
ID: {space_id}
Current Rank: {int(latest_data['rank'])}
Trending Score: {latest_data['trendingScore']:.2f}
Created At: {latest_data['createdAt'].strftime('%Y-%m-%d')}
View Space ↗
"""
chart = create_trend_chart(space_id, daily_ranks_df)
return chart, gr.HTML(value=info_text)
except Exception as e:
print(f"Error in update_display: {e}")
return None, gr.HTML(value=f"Error processing data: {str(e)}
")
def load_and_process_data():
try:
url = "https://huggingface.co/datasets/cfahlgren1/hub-stats/resolve/main/spaces.parquet"
response = requests.get(url)
df = pd.read_parquet(BytesIO(response.content))
thirty_days_ago = datetime.now() - timedelta(days=30)
df['createdAt'] = pd.to_datetime(df['createdAt'])
df = df[df['createdAt'] >= thirty_days_ago].copy()
dates = pd.date_range(start=thirty_days_ago, end=datetime.now(), freq='D')
daily_ranks = []
for date in dates:
date_data = df[df['createdAt'].dt.date <= date.date()].copy()
date_data = date_data.sort_values(['trendingScore', 'id'], ascending=[False, True])
date_data['rank'] = range(1, len(date_data) + 1)
date_data['date'] = date.date()
daily_ranks.append(
date_data[['id', 'date', 'rank', 'trendingScore', 'createdAt']]
)
daily_ranks_df = pd.concat(daily_ranks, ignore_index=True)
latest_date = daily_ranks_df['date'].max()
top_100_spaces = daily_ranks_df[
(daily_ranks_df['date'] == latest_date) &
(daily_ranks_df['rank'] <= 100)
].sort_values('rank').copy()
return daily_ranks_df, top_100_spaces
except Exception as e:
print(f"Error loading data: {e}")
return pd.DataFrame(), pd.DataFrame()
# 데이터 로드
print("Loading initial data...")
daily_ranks_df, top_100_spaces = load_and_process_data()
print("Data loaded successfully!")
# 중복 스페이스 데이터 계산
duplicates = get_duplicate_spaces(top_100_spaces)
duplicates_chart = create_duplicates_chart(duplicates)
# Gradio 인터페이스 생성
with gr.Blocks(theme=gr.themes.Soft()) as demo:
gr.Markdown("""
# HF Space Ranking Tracker(~30 Dailys)
Track, analyze, and discover trending AI applications in the Hugging Face ecosystem. Our service continuously monitors and ranks all Spaces over a 30-day period, providing detailed analytics and daily ranking changes for the top 100 performers.
""")
with gr.Tabs():
with gr.Tab("Dashboard"):
with gr.Row(variant="panel"):
with gr.Column(scale=5): # 수정된 부분
trend_plot = gr.Plot(
label="Daily Rank Trend",
container=True
)
with gr.Column(scale=5): # 수정된 부분
duplicates_plot = gr.Plot(
label="Multiple Entries Analysis",
value=duplicates_chart,
container=True
)
with gr.Row():
info_box = gr.HTML(
value="Select a space to view details
"
)
space_selection = gr.Radio(
choices=[row['id'] for _, row in top_100_spaces.iterrows()],
value=None,
visible=False
)
html_content = """
""" + "".join([
f"""
#{int(row['rank'])}
{row['id']}
Score: {row['trendingScore']:.2f}
"""
for _, row in top_100_spaces.iterrows()
]) + """