Spaces:
Sleeping
Sleeping
| import gradio as gr | |
| import regex as re | |
| import csv | |
| import pandas as pd | |
| from typing import List, Dict, Tuple, Optional | |
| import logging | |
| from datetime import datetime | |
| import os | |
| from huggingface_hub import HfApi, SpaceCard | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # Constants | |
| CSV_FILE = "repo_ids.csv" | |
| CHATBOT_SYSTEM_PROMPT = """You are a helpful AI assistant that analyzes Hugging Face repositories. | |
| Your task is to help users understand repositories, extract key information, and provide insights. | |
| Be concise, clear, and focus on the most important aspects of each repository.""" | |
| class AppState: | |
| """Simple state management for the application.""" | |
| def __init__(self): | |
| self.repo_ids: List[str] = [] | |
| self.current_repo_idx: int = 0 | |
| self.chat_history: List[Dict[str, str]] = [] | |
| def read_csv_as_text(filename: str) -> pd.DataFrame: | |
| """Read CSV file and return as DataFrame.""" | |
| try: | |
| return pd.read_csv(filename) | |
| except Exception as e: | |
| logger.error(f"Error reading CSV: {e}") | |
| return pd.DataFrame(columns=["repo id", "strength", "weaknesses", "speciality", "relevance rating"]) | |
| def write_repos_to_csv(repo_ids: List[str]) -> None: | |
| """Write repository IDs to CSV file.""" | |
| try: | |
| with open(CSV_FILE, 'w', newline='') as f: | |
| writer = csv.writer(f) | |
| writer.writerow(["repo id", "strength", "weaknesses", "speciality", "relevance rating"]) | |
| for repo_id in repo_ids: | |
| writer.writerow([repo_id, "", "", "", ""]) | |
| except Exception as e: | |
| logger.error(f"Error writing to CSV: {e}") | |
| def search_top_spaces(keyword: str, limit: int = 5) -> List[str]: | |
| """Search for repositories by keyword.""" | |
| try: | |
| api = HfApi() | |
| spaces = api.list_spaces(search=keyword, limit=limit) | |
| return [space.id for space in spaces] | |
| except Exception as e: | |
| logger.error(f"Error searching spaces: {e}") | |
| return [] | |
| def analyze_repo(repo_id: str) -> Tuple[str, str]: | |
| """Analyze a single repository.""" | |
| try: | |
| api = HfApi() | |
| space = api.get_space(repo_id) | |
| card = SpaceCard.load(repo_id) | |
| content = f""" | |
| Repository: {repo_id} | |
| Title: {card.title} | |
| Description: {card.description} | |
| Tags: {', '.join(card.tags)} | |
| """ | |
| summary = f"Analysis of {repo_id}:\n" | |
| summary += f"- Title: {card.title}\n" | |
| summary += f"- Main focus: {card.description[:200]}...\n" | |
| summary += f"- Key tags: {', '.join(card.tags[:5])}\n" | |
| return content, summary | |
| except Exception as e: | |
| logger.error(f"Error analyzing repo {repo_id}: {e}") | |
| return f"Error analyzing {repo_id}", f"Error: {str(e)}" | |
| def chat_with_user(message: str, history: List[Dict[str, str]], system_prompt: str) -> str: | |
| """Simple chat response.""" | |
| try: | |
| # For now, return a simple response | |
| return f"I understand you're asking about: {message}. How can I help you analyze this repository?" | |
| except Exception as e: | |
| logger.error(f"Error in chat: {e}") | |
| return "I apologize, but I encountered an error. Please try again." | |
| def create_ui() -> gr.Blocks: | |
| """Create a simplified Gradio interface.""" | |
| state = gr.State(AppState()) | |
| with gr.Blocks(title="Hugging Face Repo Analyzer", theme=gr.themes.Soft()) as app: | |
| gr.Markdown("# Hugging Face Repository Analyzer") | |
| with gr.Row(): | |
| with gr.Column(): | |
| # Input Section | |
| gr.Markdown("### Enter Repository Information") | |
| repo_input = gr.Textbox( | |
| label="Enter repo IDs (comma or newline separated) or keywords to search", | |
| lines=5, | |
| placeholder="Enter repository IDs or keywords to search" | |
| ) | |
| submit_btn = gr.Button("Submit", variant="primary") | |
| status = gr.Textbox(label="Status", visible=True) | |
| # Results Section | |
| df_output = gr.Dataframe( | |
| headers=["repo id", "strength", "weaknesses", "speciality", "relevance rating"], | |
| datatype=["str", "str", "str", "str", "str"] | |
| ) | |
| # Analysis Section | |
| content_output = gr.Textbox(label="Repository Content", lines=10) | |
| summary_output = gr.Textbox(label="Analysis Summary", lines=5) | |
| # Chat Section | |
| chatbot = gr.Chatbot(label="Chat with Assistant", height=400) | |
| msg = gr.Textbox(label="Message", placeholder="Ask about the repository...") | |
| with gr.Row(): | |
| send_btn = gr.Button("Send", variant="primary") | |
| clear_btn = gr.Button("Clear Chat", variant="secondary") | |
| def process_input(text: str, state: AppState) -> Tuple[pd.DataFrame, str, str, str]: | |
| """Process input and return results.""" | |
| try: | |
| # Check if input is keywords or repo IDs | |
| if any(kw in text.lower() for kw in ['search', 'find', 'look for']): | |
| # Handle as keyword search | |
| keywords = [k.strip() for k in re.split(r'[\n,]+', text) if k.strip()] | |
| repo_ids = [] | |
| for kw in keywords: | |
| repo_ids.extend(search_top_spaces(kw, limit=5)) | |
| else: | |
| # Handle as repo IDs | |
| repo_ids = [rid.strip() for rid in re.split(r'[\n,]+', text) if rid.strip()] | |
| # Remove duplicates | |
| repo_ids = list(dict.fromkeys(repo_ids)) | |
| if not repo_ids: | |
| return pd.DataFrame(), "No repositories found", "", "" | |
| # Update state and CSV | |
| state.repo_ids = repo_ids | |
| state.current_repo_idx = 0 | |
| write_repos_to_csv(repo_ids) | |
| # Get first repo analysis | |
| content, summary = analyze_repo(repo_ids[0]) | |
| return read_csv_as_text(CSV_FILE), f"Found {len(repo_ids)} repositories", content, summary | |
| except Exception as e: | |
| logger.error(f"Error processing input: {e}") | |
| return pd.DataFrame(), f"Error: {str(e)}", "", "" | |
| def send_message(message: str, history: List[Dict[str, str]], state: AppState) -> Tuple[List[Dict[str, str]], str]: | |
| """Send message to chat.""" | |
| if not message: | |
| return history, "" | |
| history.append({"role": "user", "content": message}) | |
| response = chat_with_user(message, history, CHATBOT_SYSTEM_PROMPT) | |
| history.append({"role": "assistant", "content": response}) | |
| return history, "" | |
| def clear_chat() -> Tuple[List[Dict[str, str]], str]: | |
| """Clear chat history.""" | |
| return [], "" | |
| # Event handlers | |
| submit_btn.click( | |
| fn=process_input, | |
| inputs=[repo_input, state], | |
| outputs=[df_output, status, content_output, summary_output] | |
| ) | |
| send_btn.click( | |
| fn=send_message, | |
| inputs=[msg, chatbot, state], | |
| outputs=[chatbot, msg] | |
| ) | |
| clear_btn.click( | |
| fn=clear_chat, | |
| inputs=[], | |
| outputs=[chatbot, msg] | |
| ) | |
| return app | |
| if __name__ == "__main__": | |
| app = create_ui() | |
| app.launch() | |