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',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):
"""
top_100_spaces 안에서 username/spacename 형태의 id에서 username만 떼어낸 후
(clean_id), 해당 username에 속한 여러 스페이스 점수를 합산하여 상위 20을 추출
"""
# clean_id 추출
top_100_spaces['clean_id'] = top_100_spaces['id'].apply(lambda x: x.split('/')[0])
# username별 트렌딩 스코어 합산
score_sums = top_100_spaces.groupby('clean_id')['trendingScore'].sum()
# 디버깅용 출력
print("\n=== ID별 스코어 합산 결과 (상위 20) ===")
for cid, score in score_sums.sort_values(ascending=False).head(20).items():
print(f"Clean ID: {cid}, 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=== 차트 데이터 (clean_id 단위) ===")
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=[len(df) + 0.5, 0.5], # 상위 20개 기준
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():
"""
- spaces.parquet 파일을 로드 후 30일 이내 데이터만 필터링.
- 중복 방지:
1) (선택) createdAt/ID 기준으로 중복 제거 (동일 시간대에 여러번 기록된 Space가 있으면)
2) 날짜별로 랭킹 산정 -> daily_ranks_df
3) 최종 최신 날짜 기준 Top 100 추출 후 동일 ID 중복 제거
"""
try:
url = "https://huggingface.co/datasets/cfahlgren1/hub-stats/resolve/main/spaces.parquet"
response = requests.get(url)
df = pd.read_parquet(BytesIO(response.content))
# 30일 전 시점 계산
thirty_days_ago = datetime.now() - timedelta(days=30)
df['createdAt'] = pd.to_datetime(df['createdAt'])
# 30일 내에 생성된 기록만 필터링
df = df[df['createdAt'] >= thirty_days_ago].copy()
# (선택) createdAt & id 기준 중복 제거
# 만약 동일 createdAt 시점에 동일 id가 여러 행으로 들어온 경우 가장 최신(또는 가장 높은 스코어)만 남김
df = (
df
.sort_values(['createdAt', 'trendingScore'], ascending=[True, False])
.drop_duplicates(subset=['createdAt', 'id'], keep='first')
.reset_index(drop=True)
)
# 날짜 범위 생성
dates = pd.date_range(start=thirty_days_ago, end=datetime.now(), freq='D')
daily_ranks = []
# 날짜별로 rank 계산
for date in dates:
# date 기준으로 createdAt이 date 이하인 스페이스만 추출
date_data = df[df['createdAt'].dt.date <= date.date()].copy()
# trendingScore 내림차순, id 오름차순 정렬
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)
# 최신 날짜 기준 Top 100 추출
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()
# 혹시 중복(id가 동일) 행이 있을 수 있으므로 한 번 더 제거
top_100_spaces = top_100_spaces.drop_duplicates(subset=['id'], keep='first').reset_index(drop=True)
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!")
# 중복 스페이스 데이터(= 동일 username이 여러 스페이스 운영)를 계산
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 Days)
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
"
)
# Radio 버튼은 숨겨두고, 카드 클릭으로 선택하도록 구성
space_selection = gr.Radio(
choices=[row['id'] for _, row in top_100_spaces.iterrows()],
value=None,
visible=False
)
# Top 100을 카드 형태로 표시
html_content = """
""" + "".join([
f"""
#{int(row['rank'])}
{row['id']}
Score: {row['trendingScore']:.2f}
"""
for _, row in top_100_spaces.iterrows()
]) + """