import gradio as gr import os import requests from git import Repo, GitCommandError from pathlib import Path from datetime import datetime import shutil import time import json import asyncio from typing import Dict, Optional, Tuple import logging # ========== Configuration ========== REPO_OWNER = "enricoros" REPO_NAME = "big-agi" MIAGI_FORK = "https://github.com/Ig0tU/miagi.git" WORKSPACE = Path("/tmp/issue_workspace") WORKSPACE.mkdir(exist_ok=True) HF_SPACE_API = "https://huggingface.co/api/spaces" # Hypothetical HF Space API # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ========== Enhanced Theme ========== theme = gr.themes.Default( primary_hue="emerald", secondary_hue="zinc", radius_size="lg", ).set( button_primary_background_fill="*primary_500", button_primary_background_fill_hover="*primary_400", button_primary_text_color="white", block_label_background_fill="*primary_500", block_label_text_color="white", body_background_fill="gray_100", ) # ========== State Management with Spaces Integration ========== class EnhancedIssueManager: def __init__(self): self.issues: Dict[int, dict] = {} self.last_fetched: Optional[datetime] = None self.current_issue: Optional[int] = None self.repo: Optional[Repo] = None self.space_status: Dict = {} self.progress_tracker = {} async def fetch_issues(self) -> Tuple[bool, str]: if self.last_fetched and (datetime.now() - self.last_fetched).seconds < 300: return True, "Using cached issues" try: headers = {"Authorization": f"Bearer {os.environ.get('GITHUB_TOKEN', '')}"} async with aiohttp.ClientSession() as session: async with session.get(f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/issues", headers=headers) as response: response.raise_for_status() self.issues = {issue['number']: issue for issue in await response.json()} self.last_fetched = datetime.now() return True, "Successfully fetched issues" except Exception as e: logger.error(f"Error fetching issues: {e}") return False, f"Failed to fetch issues: {str(e)}" async def check_space_status(self, issue_number: int) -> Dict: """Check if issue relates to a Hugging Face Space""" try: headers = {"Authorization": f"Bearer {os.environ.get('HF_TOKEN', '')}"} async with aiohttp.ClientSession() as session: async with session.get(f"{HF_SPACE_API}/{REPO_OWNER}/{REPO_NAME}", headers=headers) as response: if response.status == 200: space_data = await response.json() self.space_status[issue_number] = space_data return {"space_found": True, "data": space_data} return {"space_found": False} except Exception as e: logger.error(f"Space check failed: {e}") return {"space_found": False, "error": str(e)} issue_manager = EnhancedIssueManager() # ========== Core Functions with Space Handling ========== async def replicate_issue(issue_number: str) -> Tuple[Dict, str, gr.update]: try: issue_number = int(issue_number) success, fetch_message = await issue_manager.fetch_issues() if not success: return {"error": f"⚠️ {fetch_message}"}, None, None if issue_number not in issue_manager.issues: return {"error": f"❌ Issue #{issue_number} not found"}, None, None issue = issue_manager.issues[issue_number] repo_path = WORKSPACE / f"miagi-{issue_number}" issue_manager.progress_tracker[issue_number] = {"status": "initializing"} # Check Space status space_status = await issue_manager.check_space_status(issue_number) # Clone with progress try: issue_manager.progress_tracker[issue_number] = {"status": "cloning"} repo = Repo.clone_from(MIAGI_FORK, repo_path, branch="main", progress=lambda x: issue_manager.progress_tracker[issue_number].update({"progress": x})) except GitCommandError: repo = Repo(repo_path) branch_name = f"issue-{issue_number}" repo.git.checkout('-b', branch_name) # Handle Space replication if applicable issues_dir = repo_path / "issues" / str(issue_number) issues_dir.mkdir(parents=True, exist_ok=True) replication_details = { "timestamp": datetime.utcnow().isoformat(), "space_status": space_status } (issues_dir / "replication.json").write_text(json.dumps(replication_details)) issue_manager.repo = repo issue_manager.current_issue = issue_number return { "success": f"🚀 Replicated issue #{issue_number}", "space": space_status, "progress": issue_manager.progress_tracker[issue_number] }, issue["body"], gr.update(selected=1) except Exception as e: logger.error(f"Replication error: {e}") return {"error": f"⛔ Critical error: {str(e)}"}, None, None async def resolve_issue(resolution_notes: str, space_config: Optional[str] = None) -> Tuple[Dict, gr.update]: if not issue_manager.current_issue: return {"error": "⚠️ No active issue selected"}, None try: repo = issue_manager.repo issue_number = issue_manager.current_issue resolution_dir = repo.working_dir / "issues" / str(issue_number) timestamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S") resolution_data = { "notes": resolution_notes, "space_config": space_config, "resolved_at": timestamp } (resolution_dir / f"resolution_{timestamp}.json").write_text(json.dumps(resolution_data)) (resolution_dir / "RESOLVED").touch() return {"success": "🔧 Resolution saved"}, gr.update(visible=True) except Exception as e: return {"error": f"⛔ Resolution failed: {str(e)}"}, None async def submit_resolution() -> Dict: if not issue_manager.current_issue: return {"error": "⚠️ No active issue selected"} try: repo = issue_manager.repo issue_number = issue_manager.current_issue repo.git.add(all=True) commit_message = f"Resolve #{issue_number}: {issue_manager.issues[issue_number]['title']}" repo.index.commit(commit_message) origin = repo.remote(name='origin') branch_name = f"issue-{issue_number}" origin.push(refspec=f"{branch_name}:{branch_name}") # Space-specific submission if applicable if issue_manager.space_status.get(issue_number, {}).get("space_found"): await self._update_space_config(issue_number) pr_url = f"https://github.com/Ig0tU/miagi/compare/{branch_name}?expand=1" return { "success": f"🚀 Resolution submitted!\n\n[Review PR]({pr_url})", "pr_url": pr_url } except Exception as e: return {"error": f"⛔ Submission failed: {str(e)}"} async def _update_space_config(self, issue_number: int): """Update Hugging Face Space configuration""" try: headers = {"Authorization": f"Bearer {os.environ.get('HF_TOKEN', '')}"} config = {"issue_number": issue_number, "status": "resolved"} async with aiohttp.ClientSession() as session: await session.post( f"{HF_SPACE_API}/{REPO_OWNER}/{REPO_NAME}/config", json=config, headers=headers ) except Exception as e: logger.error(f"Space config update failed: {e}") # ========== Enhanced UI ========== def create_enhanced_ui(): with gr.Blocks(theme=theme, css="footer {visibility: hidden}") as app: gr.Markdown("# 🚀 Enhanced Repository Manager\n### With Hugging Face Spaces Integration") with gr.Tabs() as tabs: with gr.Tab("🔍 Investigate"): with gr.Row(): issue_input = gr.Number(label="Issue Number") replicate_btn = gr.Button("⚡ Replicate", variant="primary") status_output = gr.JSON(label="Replication Status") progress_bar = gr.Progress() with gr.Tab("🔧 Resolve"): with gr.Row(): with gr.Column(): issue_desc = gr.Markdown() resolution_input = gr.Textbox(lines=8, label="Resolution Notes") space_config = gr.Textbox(lines=2, label="Space Config (if applicable)") resolve_btn = gr.Button("💾 Save", variant="primary") with gr.Column(): file_explorer = gr.FileExplorer() with gr.Tab("🚀 Deploy"): deploy_btn = gr.Button("🔥 Submit", variant="primary") pr_output = gr.Markdown() space_status = gr.JSON(label="Space Status") # Event handlers with async support replicate_btn.click( fn=replicate_issue, inputs=issue_input, outputs=[status_output, issue_desc, tabs], _js="async (...args) => await gradioClient.run(args)" ) resolve_btn.click( fn=resolve_issue, inputs=[resolution_input, space_config], outputs=[status_output, deploy_btn] ) deploy_btn.click( fn=submit_resolution, outputs=[status_output, pr_output, space_status] ) return app # ========== Execution ========== if __name__ == "__main__": app = create_enhanced_ui() app.launch(server_port=7860, share=True)