Update app.py

app.py

@@ -1,128 +1,143 @@
-import discord
-import logging
-import os
-from huggingface_hub import InferenceClient
+import gradio as gr
 import requests
-import asyncio
-import subprocess
-from dotenv import load_dotenv
+from datetime import datetime, timedelta
+import pandas as pd
+import plotly.express as px
 
-# Load environment variables
-load_dotenv()
-
-# Set Pixabay API key
-os.environ['PIXABAY_API_KEY'] = "33492762-a28a596ec4f286f84cd328b17"
-
-# Logging setup
-logging.basicConfig(level=logging.DEBUG, format='%(asctime)s:%(levelname)s:%(name)s:%(message)s', handlers=[logging.StreamHandler()])
-
-# Intent setup
-intents = discord.Intents.default()
-intents.message_content = True
-intents.messages = True
-intents.guilds = True
-intents.guild_messages = True
-
-# Inference API client setup
-hf_client = InferenceClient("meta-llama/Meta-Llama-3.1-70B-Instruct", token=os.getenv("HF_TOKEN"))
-
-# Pixabay API setup
-PIXABAY_API_KEY = os.getenv("PIXABAY_API_KEY")
-PIXABAY_API_URL = "https://pixabay.com/api/videos/"
-
-# Specific channel ID
-SPECIFIC_CHANNEL_ID = int(os.getenv("DISCORD_CHANNEL_ID"))
-
-# Global variable to store conversation history
-conversation_history = []
-
-class MyClient(discord.Client):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.is_processing = False
-
-    async def on_ready(self):
-        logging.info(f'{self.user}로 로그인되었습니다!')
-        subprocess.Popen(["python", "web.py"])
-        logging.info("Web.py server has been started.")
-
-        # Send a guide message when the bot starts
-        channel = self.get_channel(SPECIFIC_CHANNEL_ID)
-        if channel:
-            await channel.send("찾고싶은 비디오에 대한 설명을 한 문장 단위로 입력하세요. 예) 바다에서 서핑하는 사람")
-
-    async def on_message(self, message):
-        if message.author == self.user:
-            return
-        if not self.is_message_in_specific_channel(message):
-            return
-        if self.is_processing:
-            return
-        self.is_processing = True
-        try:
-            # Extract English keywords from the meaning analysis
-            keywords = await extract_keywords(message)
-            if keywords:
-                # Search for videos using Pixabay API
-                video_urls = await search_videos(keywords)
-                if video_urls:
-                    # Create a thread with the requester and send video links
-                    await create_thread_and_send_videos(message, keywords, video_urls)
-                else:
-                    await message.channel.send(f"**{keywords}**에 대한 비디오를 찾을 수 없습니다.")
-            else:
-                await message.channel.send("키워드를 추출할 수 없습니다.")
-        finally:
-            self.is_processing = False
-
-    def is_message_in_specific_channel(self, message):
-        # Return True if the message is in the specified channel or in a thread of that channel
-        return message.channel.id == SPECIFIC_CHANNEL_ID or (
-            isinstance(message.channel, discord.Thread) and message.channel.parent_id == SPECIFIC_CHANNEL_ID
-        )
-
-async def extract_keywords(message):
-    user_input = message.content
-    system_prompt = "다음 문장의 의미에 맞는 영문 키워드를 추출하세요: "
+def get_user_spaces(token):
+    """사용자의 모든 Space 목록을 가져옵니다."""
+    headers = {"Authorization": f"Bearer {token}"}
     
-    logging.debug(f'Extracting keywords from user input: {user_input}')
-
-    messages = [{"role": "system", "content": system_prompt + user_input}]
-    logging.debug(f'Messages to be sent to the model: {messages}')
-
-    loop = asyncio.get_event_loop()
-    response = await loop.run_in_executor(None, lambda: hf_client.chat_completion(
-        messages, max_tokens=10, temperature=0.7, top_p=0.85))
-
-    # Parse Hugging Face response
-    if response.choices and response.choices[0].message:
-        keywords = response.choices[0].message['content'].strip()
-    else:
-        keywords = ""
-    logging.debug(f'Extracted keywords: {keywords}')
-    return keywords
-
-async def search_videos(keywords):
-    params = {
-        "key": PIXABAY_API_KEY,
-        "q": keywords,
-        "per_page": 40  # Get up to 40 videos
-    }
-    response = requests.get(PIXABAY_API_URL, params=params)
-    if response.status_code == 200:
-        data = response.json()
-        return [video['videos']['medium']['url'] for video in data['hits'] if 'videos' in video]
-    logging.error(f"Pixabay API error: {response.status_code}, {response.text}")
-    return None
-
-async def create_thread_and_send_videos(message, keywords, video_urls):
-    # Create a thread
-    thread = await message.channel.create_thread(name=f"{message.author.name}의 검색 결과", message=message)
-    message_content = f"**{keywords}**에 대한 비디오 {len(video_urls)}개를 찾았습니다:"
-    await thread.send(message_content)
-    for url in video_urls:
-        await thread.send(url)
+    try:
+        # 사용자 정보 가져오기
+        user_info = requests.get("https://huggingface.co/api/whoami", headers=headers)
+        user_info.raise_for_status()
+        username = user_info.json().get('name')
+        
+        if not username:
+            return None, "사용자 정보를 가져올 수 없습니다."
+            
+        # 사용자의 Space 목록 가져오기
+        spaces_url = f"https://huggingface.co/api/spaces/{username}"
+        spaces_response = requests.get(spaces_url, headers=headers)
+        spaces_response.raise_for_status()
+        
+        return username, spaces_response.json()
+        
+    except requests.exceptions.RequestException as e:
+        return None, f"API 요청 실패: {str(e)}"
+
+def get_space_visitors(owner, space_name, token):
+    """특정 Space의 방문자 통계를 가져옵니다."""
+    url = f"https://huggingface.co/api/spaces/{owner}/{space_name}/metrics"
+    headers = {"Authorization": f"Bearer {token}"}
+    
+    try:
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()
+        
+        # 최근 7일간의 데이터만 필터링
+        today = datetime.now()
+        week_ago = today - timedelta(days=7)
+        
+        daily_visits = {}
+        total_visits = 0
+        
+        for record in response.json():
+            date = datetime.fromtimestamp(record['timestamp'])
+            if date >= week_ago:
+                date_str = date.strftime('%Y-%m-%d')
+                visits = record.get('visits', 0)
+                daily_visits[date_str] = daily_visits.get(date_str, 0) + visits
+                total_visits += visits
+                
+        return {
+            'name': space_name,
+            'total_visits': total_visits,
+            'daily_visits': daily_visits
+        }
+        
+    except requests.exceptions.RequestException:
+        return None
+
+def analyze_spaces(token):
+    """모든 Space의 방문자 통계를 분석합니다."""
+    if not token:
+        return "API 토큰을 입력해주세요.", None, None
+    
+    # 사용자 정보와 Space 목록 가져오기
+    username, spaces_data = get_user_spaces(token)
+    
+    if not username:
+        return f"오류: {spaces_data}", None, None
+    
+    if not isinstance(spaces_data, list):
+        return "Space 목록을 가져올 수 없습니다.", None, None
+    
+    # 각 Space의 방문자 통계 수집
+    all_stats = []
+    daily_data = []
+    
+    for space in spaces_data:
+        space_name = space.get('id')
+        if space_name:
+            stats = get_space_visitors(username, space_name, token)
+            if stats:
+                all_stats.append(stats)
+                for date, visits in stats['daily_visits'].items():
+                    daily_data.append({
+                        'date': date,
+                        'space': space_name,
+                        'visits': visits
+                    })
+    
+    if not all_stats:
+        return "방문자 통계를 가져올 수 없습니다.", None, None
+    
+    # 결과 텍스트 생성
+    result_text = f"계정: {username}\n\n"
+    for stats in sorted(all_stats, key=lambda x: x['total_visits'], reverse=True):
+        result_text += f"Space: {stats['name']}\n"
+        result_text += f"총 방문자 수: {stats['total_visits']}명\n\n"
+    
+    # 그래프 생성
+    if daily_data:
+        df = pd.DataFrame(daily_data)
+        fig = px.line(df, x='date', y='visits', color='space',
+                     title=f'Space별 일일 방문자 수',
+                     labels={'date': '날짜', 'visits': '방문자 수', 'space': 'Space 이름'})
+        
+        # 막대 그래프 생성
+        summary_df = pd.DataFrame([(s['name'], s['total_visits']) for s in all_stats],
+                                columns=['space', 'total_visits'])
+        summary_fig = px.bar(summary_df, x='space', y='total_visits',
+                           title='Space별 총 방문자 수',
+                           labels={'space': 'Space 이름', 'total_visits': '총 방문자 수'})
+        
+        return result_text, fig, summary_fig
+    
+    return result_text, None, None
+
+# Gradio 인터페이스 생성
+with gr.Blocks() as app:
+    gr.Markdown("""
+    # Hugging Face Space 방문자 통계
+    API 토큰을 입력하면 모든 Space의 방문자 통계를 확인할 수 있습니다.
+    """)
+    
+    with gr.Row():
+        token_input = gr.Textbox(label="API 토큰", type="password")
+        submit_btn = gr.Button("통계 확인")
+    
+    stats_output = gr.Textbox(label="통계 정보", lines=10)
+    plot_output = gr.Plot(label="일별 방문자 수")
+    summary_plot = gr.Plot(label="총 방문자 수")
+    
+    submit_btn.click(
+        analyze_spaces,
+        inputs=[token_input],
+        outputs=[stats_output, plot_output, summary_plot]
+    )
 
 if __name__ == "__main__":
-    discord_client = MyClient(intents=intents)
-    discord_client.run(os.getenv('DISCORD_TOKEN'))
+    app.launch()