import gradio as gr import re import json import os import tempfile import shlex from huggingface_hub import HfApi from build_logic import ( _get_api_token as build_logic_get_api_token, whoami as build_logic_whoami, list_space_files_for_browsing, get_space_repository_info, get_space_file_content, update_space_file, parse_markdown as build_logic_parse_markdown, delete_space_file as build_logic_delete_space_file, get_space_runtime_status, apply_staged_file_changes, # Renamed to indicate it only handles files duplicate_space as build_logic_duplicate_space, list_user_spaces as build_logic_list_user_spaces, build_logic_set_space_privacy, build_logic_delete_space, build_logic_create_pull_request, build_logic_add_comment, build_logic_like_space ) from model_logic import ( get_available_providers, get_models_for_provider, get_default_model_for_provider, generate_stream ) backtick = chr(96) bbb = f'{backtick}{backtick}{backtick}' parsed_code_blocks_state_cache = [] BOT_ROLE_NAME = "assistant" DEFAULT_SYSTEM_PROMPT = f"""You are an expert AI programmer and Hugging Face assistant. Your role is to generate code and file structures based on user requests, or to modify existing code provided by the user. You operate in the context of a Hugging Face Space. You can propose file changes and trigger specific actions. **File and Code Formatting:** When you provide NEW code for a file, or MODIFIED code for an existing file, use the following format exactly: ### File: path/to/filename.ext (You can add a short, optional, parenthesized description after the filename on the SAME line) {bbb}language # Your full code here {bbb} If the file is binary, or you cannot show its content, use this format: ### File: path/to/binaryfile.ext [Binary file - approximate_size bytes] When you provide a project file structure, use this format (for informational purposes, not parsed for changes): ## File Structure {bbb} 📁 Root 📄 file1.py 📁 subfolder 📄 file2.js {bbb} **Instructions and Rules:** - The role name for your responses in the chat history must be '{BOT_ROLE_NAME}'. - Adhere strictly to these formatting instructions. - If you update a file, provide the FULL file content again under the same filename. - Only the latest version of each file mentioned throughout the chat will be used for the final output. The system will merge your changes with the prior state before applying. **Hugging Face Space Actions:** To perform direct actions on the Hugging Face Space, use the `### HF_ACTION: COMMAND arguments...` command on a single line. The system will parse these actions from your response and present them to the user for confirmation before executing. Available commands: - `CREATE_SPACE / --sdk --private `: Creates a new, empty space. SDK can be gradio, streamlit, docker, or static. Private is optional and defaults to false. Use this ONLY when the user explicitly asks to create a *new* space with a specific name. When using this action, also provide the initial file structure (e.g., app.py, README.md) using `### File:` blocks in the same response. The system will apply these files to the new space. - `DUPLICATE_SPACE / --private `: Duplicates the *currently loaded* space to a new space. Private is optional and defaults to false. This action is destructive if the target space already exists. Use this ONLY when the user explicitly asks to duplicate the space. If this action is staged, it will be the *only* action applied in that changeset. - `DELETE_FILE path/to/file.ext`: Deletes a specific file from the current space. - `SET_PRIVATE `: Sets the privacy for the current space. - `DELETE_SPACE`: Deletes the entire current space. THIS IS PERMANENT AND REQUIRES CAUTION. Only use this if the user explicitly and clearly asks to delete the space. - `CREATE_PR --title "..." --body "..."`: Creates a Pull Request from the current space's main branch to the target repository (e.g., a model or dataset repo) main branch. Requires `--title` and `--body` arguments (quoted strings). - `ADD_COMMENT "Your comment text"`: Adds a comment to the currently loaded space. Requires a quoted string comment text. - `LIKE_SPACE`: Likes the currently loaded space. You can issue multiple file updates and action commands in a single response. The system will process all of them into a single change plan. Note that the `DUPLICATE_SPACE` command, if present, will override any other file or space actions in the same response. Actions other than file changes are executed sequentially before any file changes are committed. **Current Space Context:** You will be provided with the current state of the files in the Space the user is interacting with. Use this information to understand the current project structure and content before proposing changes or actions. This context will appear after the user's message, starting with "## Current Space Context:". Do NOT include this context in your response. Only generate your response based on the user's request and the formatting rules above. If no code or actions are requested, respond conversationally and help the user understand the Space Commander's capabilities. """ def escape_html_for_markdown(text): if not isinstance(text, str): return "" return text.replace("&", "&").replace("<", "<").replace(">", ">") def _infer_lang_from_filename(filename): if not filename: return "plaintext" if '.' in filename: ext = filename.split('.')[-1].lower() mapping = { 'py': 'python', 'js': 'javascript', 'ts': 'typescript', 'jsx': 'javascript', 'tsx': 'typescript', 'html': 'html', 'htm': 'html', 'css': 'css', 'scss': 'scss', 'sass': 'sass', 'less': 'less', 'json': 'json', 'xml': 'xml', 'yaml': 'yaml', 'yml': 'yaml', 'toml': 'toml', 'md': 'markdown', 'rst': 'rst', 'sh': 'bash', 'bash': 'bash', 'zsh': 'bash', 'bat': 'batch', 'cmd': 'batch', 'ps1': 'powershell', 'c': 'c', 'h': 'c', 'cpp': 'cpp', 'hpp': 'cpp', 'cs': 'csharp', 'java': 'java', 'rb': 'ruby', 'php': 'php', 'go': 'go', 'rs': 'rust', 'swift': 'swift', 'kt': 'kotlin', 'kts': 'kotlin', 'sql': 'sql', 'dockerfile': 'docker', 'tf': 'terraform', 'hcl': 'terraform', 'txt': 'plaintext', 'log': 'plaintext', 'ini': 'ini', 'conf': 'plaintext', 'cfg': 'plaintext', 'csv': 'plaintext', 'tsv': 'plaintext', 'err': 'plaintext', '.env': 'plaintext', '.gitignore': 'plaintext', '.npmrc': 'plaintext', '.gitattributes': 'plaintext', 'makefile': 'makefile', } return mapping.get(ext, "plaintext") base_filename = os.path.basename(filename) if base_filename == 'Dockerfile': return 'docker' if base_filename == 'Makefile': return 'makefile' if base_filename.startswith('.'): return 'plaintext' return "plaintext" def _clean_filename(filename_line_content): text = filename_line_content.strip() text = re.sub(r'[`\*_#]+', '', text).strip() text = re.split(r'\s*\(', text, 1)[0].strip() text = text.strip('\'":;,') text = text.lstrip('/') return text def _parse_and_update_state_cache(latest_bot_message_content, current_files_state): current_files_dict = {f["filename"]: f.copy() for f in current_files_state if not f.get("is_structure_block")} structure_block_state = next((b for b in current_files_state if b.get("is_structure_block")), None) content = latest_bot_message_content or "" file_pattern = re.compile(r"### File:\s*(?P[^\n]+)\n(?:```(?P[\w\.\-\+]*)\n(?P[\s\S]*?)\n```|(?P\[Binary file(?: - [^\]]+)?\]))", re.MULTILINE) structure_pattern = re.compile(r"## File Structure\n```(?:(?P[\w.-]*)\n)?(?P[\s\S]*?)\n```", re.MULTILINE) structure_match = structure_pattern.search(content) if structure_match: structure_block_state = {"filename": "File Structure (from AI)", "language": structure_match.group("struct_lang") or "plaintext", "code": structure_match.group("structure_code").strip(), "is_binary": False, "is_structure_block": True} current_message_proposed_filenames = [] for match in file_pattern.finditer(content): filename = _clean_filename(match.group("filename_line")) if not filename: continue lang, code_block, binary_msg = match.group("lang"), match.group("code"), match.group("binary_msg") item_data = {"filename": filename, "is_binary": False, "is_structure_block": False} if code_block is not None: item_data["code"] = code_block.strip() item_data["language"] = (lang.strip().lower() if lang else _infer_lang_from_filename(filename)) item_data["is_binary"] = False elif binary_msg is not None: item_data["code"] = binary_msg.strip() item_data["language"] = "binary" item_data["is_binary"] = True else: continue current_files_dict[filename] = item_data current_message_proposed_filenames.append(filename) updated_parsed_blocks = list(current_files_dict.values()) if structure_block_state: updated_parsed_blocks.insert(0, structure_block_state) updated_parsed_blocks.sort(key=lambda b: (0, b["filename"]) if b.get("is_structure_block") else (1, b["filename"])) return updated_parsed_blocks, current_message_proposed_filenames def _export_selected_logic(selected_filenames, space_line_name_for_md, parsed_blocks_for_export): results = {"output_str": "", "error_message": None, "download_filepath": None} file_blocks_for_export = [b for b in parsed_blocks_for_export if not b.get("is_structure_block")] all_filenames_in_state = sorted(list(set(b["filename"] for b in file_blocks_for_export))) output_lines = [f"# Space: {space_line_name_for_md}"] structure_block = next((b for b in parsed_blocks_for_export if b.get("is_structure_block")), None) if structure_block: output_lines.extend(["## File Structure", bbb, structure_block["code"].strip(), bbb, ""]) else: output_lines.extend(["## File Structure", bbb, "📁 Root"]) if all_filenames_in_state: for fname in all_filenames_in_state: output_lines.append(f" 📄 {fname}") output_lines.extend([bbb, ""]) output_lines.append("Below are the contents of all files in the space:\n") blocks_to_export_content = sorted([b for b in file_blocks_for_export if not selected_filenames or b["filename"] in selected_filenames], key=lambda b: b["filename"]) exported_content_count = 0 for block in blocks_to_export_content: output_lines.append(f"### File: {block['filename']}") content = block.get('code', '') if block.get('is_binary') or content.startswith(("[Binary file", "[Error loading content:", "[Binary or Skipped file]")): output_lines.append(content) else: lang = block.get('language', 'plaintext') or 'plaintext' output_lines.extend([f"{bbb}{lang}", content, bbb]) output_lines.append("") exported_content_count += 1 if not exported_content_count: if selected_filenames: output_lines.append("*No selected files have editable content in the state.*") elif not all_filenames_in_state: output_lines.append("*No files in state to list structure or export.*") final_output_str = "\n".join(output_lines) results["output_str"] = final_output_str try: with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".md", encoding='utf-8') as tmpfile: tmpfile.write(final_output_str); results["download_filepath"] = tmpfile.name except Exception as e: print(f"Error creating temp file for download: {e}") results["error_message"] = "Could not prepare file for download." results["download_filepath"] = None return results def _convert_gr_history_to_api_messages(system_prompt, gr_history, current_user_message=None): messages = [{"role": "system", "content": system_prompt}] if system_prompt else [] for user_msg, bot_msg in gr_history: if user_msg: messages.append({"role": "user", "content": user_msg}) if bot_msg and isinstance(bot_msg, str): messages.append({"role": BOT_ROLE_NAME, "content": bot_msg}) if current_user_message: messages.append({"role": "user", "content": current_user_message}) return messages def _generate_ui_outputs_from_cache(owner, space_name): global parsed_code_blocks_state_cache preview_md_val = "*No files in cache to display.*" space_line_name = f"{owner}/{space_name}" if owner and space_name else (owner or space_name or "your-space") export_result = _export_selected_logic(None, space_line_name, parsed_code_blocks_state_cache) formatted_md_val = export_result["output_str"] download_file = export_result["download_filepath"] formatted_md_val = formatted_md_val or "*Load or define a Space to see its Markdown structure.*" if parsed_code_blocks_state_cache: preview_md_lines = ["## Detected/Updated Files & Content (Latest Versions):"] file_blocks_in_cache = sorted([b for b in parsed_code_blocks_state_cache if not b.get("is_structure_block")], key=lambda b: b["filename"]) structure_block_in_cache = next((b for b in parsed_code_blocks_state_cache if b.get("is_structure_block")), None) if structure_block_in_cache: preview_md_lines.append(f"\n----\n**File Structure:** (from AI)\n{bbb}\n{escape_html_for_markdown(structure_block_in_cache['code'].strip())}\n{bbb}\n") if file_blocks_in_cache: for block in file_blocks_in_cache: preview_md_lines.append(f"\n----\n**File:** `{escape_html_for_markdown(block['filename'])}`") if block.get('is_binary'): preview_md_lines.append(f" (Binary File or Skipped)\n") else: preview_md_lines.append(f" (Language: `{block['language']}`)\n") content = block.get('code', '') if block.get('is_binary') or content.startswith(("[Binary file", "[Error loading content:", "[Binary or Skipped file]")): preview_md_lines.append(f"\n`{escape_html_for_markdown(content.strip())}`\n") else: lang = block.get('language', 'plaintext') or 'plaintext' preview_md_lines.append(f"\n{bbb}{lang}\n{content.strip()}\n{bbb}\n") preview_md_val = "\n".join(preview_md_lines) return formatted_md_val, preview_md_val, gr.update(value=download_file, interactive=download_file is not None) def generate_and_stage_changes(ai_response_content, current_files_state, hf_owner_name, hf_repo_name): changeset = [] current_files_dict = {f["filename"]: f for f in current_files_state if not f.get("is_structure_block")} ai_parsed_md = build_logic_parse_markdown(ai_response_content) ai_proposed_files_list = ai_parsed_md.get("files", []) ai_proposed_files_dict = {f["path"]: f for f in ai_proposed_files_list} action_pattern = re.compile(r"### HF_ACTION:\s*(?P[^\n]+)", re.MULTILINE) potential_actions = [] for match in action_pattern.finditer(ai_response_content): try: cmd_parts = shlex.split(match.group("command_line").strip()) if not cmd_parts: continue command, args = cmd_parts[0].upper(), cmd_parts[1:] potential_actions.append({"command": command, "args": args, "raw_line": match.group("command_line").strip()}) except Exception as e: print(f"Error parsing HF_ACTION line '{match.group('command_line').strip()}': {e}") duplicate_action = next((act for act in potential_actions if act["command"] == "DUPLICATE_SPACE"), None) if duplicate_action: cmd_parts = duplicate_action["args"] if cmd_parts: target_repo_id = cmd_parts[0] private = False if '--private' in cmd_parts: try: private_str = cmd_parts[cmd_parts.index('--private') + 1].lower() except IndexError: print("Warning: DUPLICATE_SPACE --private requires an argument.") else: private = private_str == 'true' source_repo_id = f"{hf_owner_name}/{hf_repo_name}" if hf_owner_name and hf_repo_name else None if source_repo_id: changeset.append({ "type": "DUPLICATE_SPACE", "source_repo_id": source_repo_id, "target_repo_id": target_repo_id, "private": private }) print(f"Staged exclusive DUPLICATE_SPACE action from {source_repo_id} to {target_repo_id}") else: print(f"Warning: Cannot stage DUPLICATE_SPACE action, no Space currently loaded.") changeset.append({"type": "Error", "message": "Cannot duplicate space, no Space currently loaded."}) else: print(f"Warning: DUPLICATE_SPACE action requires a target repo_id.") changeset.append({"type": "Error", "message": "DUPLICATE_SPACE action requires a target repo_id (/)."}) if changeset: md_summary = ["### 📋 Proposed Changes Plan (Exclusive Action)\n"] if changeset[0]["type"] == "DUPLICATE_SPACE": change = changeset[0] md_summary.append(f"- **📂 Duplicate Space:** Duplicate `{change.get('source_repo_id', '...')}` to `{change.get('target_repo_id', '...')}` (Private: {change.get('private', False)})") md_summary.append("\n**Warning:** This will overwrite the target space if it exists. No other file or space actions proposed by the AI in this turn will be applied.") else: md_summary.append(f"- **Error staging DUPLICATE_SPACE:** {changeset[0].get('message', 'Unknown error')}") md_summary.append("\nNo changes were staged due to the error in the DUPLICATE_SPACE command.") changeset = [] return changeset, "\n".join(md_summary) for action in potential_actions: command, args = action["command"], action["args"] if command == "CREATE_SPACE" and args: repo_id = args[0] sdk = "gradio" private = False if '--sdk' in args: try: sdk = args[args.index('--sdk') + 1] except IndexError: print("Warning: CREATE_SPACE --sdk requires an argument.") if '--private' in args: try: private_str = args[args.index('--private') + 1].lower() except IndexError: print("Warning: CREATE_SPACE --private requires an argument.") else: private = private_str == 'true' changeset.append({"type": "CREATE_SPACE", "repo_id": repo_id, "sdk": sdk, "private": private}) print(f"Staged CREATE_SPACE action for {repo_id}") elif command == "DELETE_FILE" and args: file_path = args[0] changeset.append({"type": "DELETE_FILE", "path": file_path}) print(f"Staged DELETE_FILE action for {file_path}") elif command == "SET_PRIVATE" and args: private = args[0].lower() == 'true' changeset.append({"type": "SET_PRIVACY", "private": private, "repo_id": f"{hf_owner_name}/{hf_repo_name}"}) print(f"Staged SET_PRIVACY action for {hf_owner_name}/{hf_repo_name} to {private}") elif command == "DELETE_SPACE": changeset.append({"type": "DELETE_SPACE", "owner": hf_owner_name, "space_name": hf_repo_name}) print(f"Staged DELETE_SPACE action for {hf_owner_name}/{hf_repo_name}") elif command == "CREATE_PR" and args: # Expecting: CREATE_PR --title "..." --body "..." target_repo_id = None title = "" body = "" try: target_repo_id = args[0] arg_string = shlex.join(args[1:]) # Rejoin remaining args as a string to parse options title_match = re.search(r'--title\s+"([^"]*)"', arg_string) if title_match: title = title_match.group(1) else: print("Warning: CREATE_PR missing --title argument.") body_match = re.search(r'--body\s+"([^"]*)"', arg_string) if body_match: body = body_match.group(1) else: print("Warning: CREATE_PR missing --body argument.") if target_repo_id and title: # Require target and title changeset.append({"type": "CREATE_PR", "source_repo_id": f"{hf_owner_name}/{hf_repo_name}", "target_repo_id": target_repo_id, "title": title, "body": body}) print(f"Staged CREATE_PR action to {target_repo_id}") else: print("Warning: CREATE_PR requires target_repo_id and --title.") changeset.append({"type": "Error", "message": "CREATE_PR action requires target_repo_id and --title arguments."}) except Exception as e: print(f"Error parsing CREATE_PR arguments: {e}") changeset.append({"type": "Error", "message": f"Error parsing CREATE_PR arguments: {e}"}) elif command == "ADD_COMMENT" and args: # Expecting: ADD_COMMENT "Your comment text" comment_text = None try: # Attempt to parse the rest of the line as a single quoted string # Rejoin args and then try to find the quoted string comment_string = shlex.join(args) comment_match = re.match(r'^"([^"]*)"$', comment_string) # Look for the whole string being a quoted value if comment_match: comment_text = comment_match.group(1) else: # Fallback: just use the raw args joined by space if not a perfect quoted string comment_text = " ".join(args) print(f"Warning: ADD_COMMENT argument was not a single quoted string, using raw args: '{comment_text}'") if comment_text: changeset.append({"type": "ADD_COMMENT", "repo_id": f"{hf_owner_name}/{hf_repo_name}", "comment": comment_text}) print(f"Staged ADD_COMMENT action on {hf_owner_name}/{hf_repo_name}") else: print("Warning: ADD_COMMENT requires comment text.") changeset.append({"type": "Error", "message": "ADD_COMMENT action requires comment text (in quotes)."}) except Exception as e: print(f"Error parsing ADD_COMMENT arguments: {e}") changeset.append({"type": "Error", "message": f"Error parsing ADD_COMMENT arguments: {e}"}) elif command == "LIKE_SPACE": # No args expected if hf_owner_name and hf_repo_name: changeset.append({"type": "LIKE_SPACE", "repo_id": f"{hf_owner_name}/{hf_repo_name}"}) print(f"Staged LIKE_SPACE action on {hf_owner_name}/{hf_repo_name}") else: print(f"Warning: Cannot stage LIKE_SPACE action, no Space currently loaded.") changeset.append({"type": "Error", "message": "Cannot like space, no Space currently loaded."}) # Note: DUPLICATE_SPACE is handled before this loop for file_info in ai_proposed_files_list: filename = file_info["path"] proposed_content = file_info["content"] if filename in current_files_dict: current_content = current_files_dict[filename]["code"] is_proposed_placeholder = proposed_content.startswith(("[Binary file", "[Error loading content:", "[Binary or Skipped file]")) if not is_proposed_placeholder and proposed_content != current_content: lang = current_files_dict[filename].get("language") or _infer_lang_from_filename(filename) changeset.append({"type": "UPDATE_FILE", "path": filename, "content": proposed_content, "lang": lang}) elif is_proposed_placeholder: print(f"Skipping staging update for {filename}: Proposed content is a placeholder.") else: proposed_content = file_info["content"] if not (proposed_content.startswith("[Binary file") or proposed_content.startswith("[Error loading content:") or proposed_content.startswith("[Binary or Skipped file]")): lang = _infer_lang_from_filename(filename) changeset.append({"type": "CREATE_FILE", "path": filename, "content": proposed_content, "lang": lang}) else: print(f"Skipping staging create for {filename}: Proposed content is a placeholder.") if not changeset: md_summary = ["### 📋 Proposed Changes Plan", "\nThe AI did not propose any specific changes to files or the space.\n"] else: md_summary = ["### 📋 Proposed Changes Plan\n"] md_summary.append("The AI has proposed the following changes. Please review and confirm.") file_changes = [c for c in changeset if c['type'] in ['CREATE_FILE', 'UPDATE_FILE', 'DELETE_FILE']] space_actions = [c for c in changeset if c['type'] not in ['CREATE_FILE', 'UPDATE_FILE', 'DELETE_FILE']] if space_actions: md_summary.append("\n**Space Actions:**") for change in space_actions: if change["type"] == "CREATE_SPACE": md_summary.append(f"- **🚀 Create New Space:** `{change.get('repo_id', '...')}` (SDK: {change.get('sdk', 'gradio')}, Private: {change.get('private', False)})") elif change["type"] == "SET_PRIVACY": md_summary.append(f"- **🔒 Set Privacy:** Set `{change.get('repo_id', '...')}` to `private={change.get('private', False)}`") elif change["type"] == "DELETE_SPACE": md_summary.append(f"- **💥 DELETE ENTIRE SPACE:** `{change.get('owner', '...')}/{change.get('space_name', '...')}` **(DESTRUCTIVE ACTION)**") elif change["type"] == "CREATE_PR": md_summary.append(f"- **🤝 Create Pull Request:** From `{change.get('source_repo_id', '...')}` to `{change.get('target_repo_id', '...')}`") md_summary.append(f" - Title: `{change.get('title', '...')}`") if change.get('body'): md_summary.append(f" - Body: `{change.get('body', '...')}`") elif change["type"] == "ADD_COMMENT": md_summary.append(f"- **💬 Add Comment:** On `{change.get('repo_id', '...')}`") md_summary.append(f" - Comment: `{change.get('comment', '...')[:100]}{'...' if len(change.get('comment', '')) > 100 else ''}`") # Truncate long comments elif change["type"] == "LIKE_SPACE": md_summary.append(f"- **👍 Like Space:** `{change.get('repo_id', '...')}`") elif change["type"] == "Error": md_summary.append(f"- **❌ Staging Error:** {change.get('message', 'Unknown error')}") md_summary.append("") if file_changes: md_summary.append("**File Changes:**") for change in file_changes: if change["type"] == "CREATE_FILE": md_summary.append(f"- **➕ Create File:** `{change['path']}`") elif change["type"] == "UPDATE_FILE": md_summary.append(f"- **🔄 Update File:** `{change['path']}`") elif change["type"] == "DELETE_FILE": md_summary.append(f"- **➖ Delete File:** `{change['path']}`") return changeset, "\n".join(md_summary) def handle_chat_submit(user_message, chat_history, hf_api_key_input, provider_api_key_input, provider_select, model_select, system_prompt, hf_owner_name, hf_repo_name): global parsed_code_blocks_state_cache _chat_msg_in = user_message _chat_hist = list(chat_history) _chat_hist.append((user_message, None)) yield ( "", _chat_hist, "Initializing...", gr.update(), gr.update(), gr.update(interactive=False), [], gr.update(value="*No changes proposed.*"), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False) ) current_sys_prompt = system_prompt.strip() or DEFAULT_SYSTEM_PROMPT space_id_for_context = f"{hf_owner_name}/{hf_repo_name}" if hf_owner_name and hf_repo_name else "your-space" export_result = _export_selected_logic(None, space_id_for_context, parsed_code_blocks_state_cache) current_files_context_md = export_result['output_str'] or "*No files currently loaded or defined.*" current_files_context = f"\n\n## Current Space Context: {space_id_for_context}\n{current_files_context_md}" user_message_with_context = user_message.strip() + current_files_context api_msgs = _convert_gr_history_to_api_messages(current_sys_prompt, _chat_hist[:-1], user_message_with_context) full_bot_response_content = "" try: streamer = generate_stream(provider_select, model_select, provider_api_key_input, api_msgs) for chunk in streamer: if chunk is None: continue full_bot_response_content += str(chunk) _chat_hist[-1] = (user_message, full_bot_response_content) yield ( gr.update(), _chat_hist, f"Streaming from {model_select}...", gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update() ) if full_bot_response_content.startswith("Error:") or full_bot_response_content.startswith("API HTTP Error"): _status = full_bot_response_content yield ( gr.update(), _chat_hist, _status, gr.update(), gr.update(), gr.update(), [], "*Error occurred, changes plan cleared.*", gr.update(visible=False), gr.update(visible=False), gr.update(visible=False) ) return parsed_code_blocks_state_cache, proposed_filenames_in_turn = _parse_and_update_state_cache(full_bot_response_content, parsed_code_blocks_state_cache) _formatted, _detected, _download = _generate_ui_outputs_from_cache(hf_owner_name, hf_repo_name) staged_changeset, summary_md = generate_and_stage_changes(full_bot_response_content, parsed_code_blocks_state_cache, hf_owner_name, hf_repo_name) if not staged_changeset: _status = summary_md yield ( gr.update(), _chat_hist, _status, _detected, _formatted, _download, [], summary_md, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False) ) else: _status = "Change plan generated. Please review and confirm below." yield ( gr.update(), _chat_hist, _status, _detected, _formatted, _download, staged_changeset, summary_md, gr.update(visible=True), gr.update(visible=True), gr.update(visible=True) ) except Exception as e: error_msg = f"An unexpected error occurred during chat submission: {e}" print(f"Error in handle_chat_submit: {e}") import traceback traceback.print_exc() if _chat_hist: try: idx = next(i for i in reversed(range(len(_chat_hist))) if _chat_hist[i] and _chat_hist[i][0] == user_message) current_bot_response = _chat_hist[idx][1] or "" _chat_hist[idx] = (user_message, (current_bot_response + "\n\n" if current_bot_response else "") + error_msg) except (StopIteration, IndexError): _chat_hist.append((user_message, error_msg)) _formatted, _detected, _download = _generate_ui_outputs_from_cache(hf_owner_name, hf_repo_name) yield ( gr.update(), _chat_hist, error_msg, _detected, _formatted, _download, [], "*Error occurred, changes plan cleared.*", gr.update(visible=False), gr.update(visible=False), gr.update(visible=False) ) def handle_confirm_changes(hf_api_key, owner_name, space_name, changeset): global parsed_code_blocks_state_cache _status = "Applying changes..." # Yield initial status, hide confirm UI, clear summary yield _status, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(value="*Applying changes...*") # Include outputs for potential space load/update yield gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update() # Dummy yields to match structure if not changeset: return "No changes to apply.", gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), gr.update(value="No changes were staged.") # Check if the first action is an exclusive action like DUPLICATE_SPACE first_action = changeset[0] if changeset else None is_exclusive_duplicate = first_action and first_action.get('type') == 'DUPLICATE_SPACE' is_exclusive_delete = first_action and first_action.get('type') == 'DELETE_SPACE' if is_exclusive_duplicate: change = first_action if change.get("source_repo_id") and change.get("target_repo_id"): status_message = build_logic_duplicate_space( hf_api_key=hf_api_key, source_repo_id=change["source_repo_id"], target_repo_id=change["target_repo_id"], private=change.get("private", False) ) # After duplication, attempt to load the *new* space _status_reload = f"{status_message} | Attempting to load the new Space [{change['target_repo_id']}]..." yield gr.update(value=_status_reload) # Update status # Extract new owner and space name from target_repo_id target_repo_id = change['target_repo_id'] new_owner, new_space_name = target_repo_id.split('/', 1) if '/' in target_repo_id else (None, target_repo_id) # If owner was not explicitly provided in target_repo_id, try to get it from the token if not new_owner: token, token_err = build_logic_get_api_token(hf_api_key) if token_err: print(f"Error getting token for new owner determination: {token_err}") new_owner = None # Cannot auto-detect else: try: user_info = build_logic_whoami(token=token); new_owner = user_info.get('name') except Exception as e: print(f"Error auto-detecting owner from token for new space: {e}") new_owner = None # Trigger the load logic for the new space if owner and name are available if new_owner and new_space_name: sdk, file_list, err_list = get_space_repository_info(hf_api_key, new_space_name, new_owner) if err_list: reload_error = f"Error reloading file list after duplication: {err_list}" parsed_code_blocks_state_cache = [] else: loaded_files = [] for file_path in file_list: content, err_get = get_space_file_content(hf_api_key, new_space_name, new_owner, file_path) lang = _infer_lang_from_filename(file_path) is_binary = lang == "binary" or (err_get is not None) code = f"[Error loading content: {err_get}]" if err_get else (content or "") loaded_files.append({"filename": file_path, "code": code, "language": lang, "is_binary": is_binary, "is_structure_block": False}) parsed_code_blocks_state_cache = loaded_files _formatted, _detected, _download = _generate_ui_outputs_from_cache(new_owner, new_space_name) final_overall_status = status_message + (f" | Reload Status: {reload_error}" if reload_error else f" | Reload Status: New Space [{new_owner}/{new_space_name}] state refreshed.") owner_update = gr.update(value=new_owner) space_update = gr.update(value=new_space_name) file_browser_update = gr.update(visible=True, choices=sorted([f["filename"] for f in parsed_code_blocks_state_cache if not f.get("is_structure_block")] or []), value=None) if new_owner and new_space_name: sub_owner = re.sub(r'[^a-z0-9\-]+', '-', new_owner.lower()).strip('-') or 'owner' sub_repo = re.sub(r'[^a-z0-9\-]+', '-', new_space_name.lower()).strip('-') or 'space' iframe_url = f"https://{sub_owner}-{sub_repo}{'.static.hf.space' if sdk == 'static' else '.hf.space'}" iframe_update = gr.update(value=f'', visible=True) else: iframe_update = gr.update(value=None, visible=False) # Fetch runtime status for the new space runtime_status_md = handle_refresh_space_status(hf_api_key, new_owner, new_space_name) # Final yield after loading the new space yield ( gr.update(value=final_overall_status), gr.update(value=_formatted), gr.update(value=_detected), _download, # Status, markdown, preview, download gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), # Hide confirm UI [], gr.update(value="*No changes proposed.*"), # Clear changeset state and display owner_update, space_update, file_browser_update, iframe_update, # Update space/owner fields, file browser, iframe gr.update(value=runtime_status_md) # Update runtime status ) else: reload_error = "Cannot load new Space state: Owner or Space Name missing after duplication." _formatted, _detected, _download = _generate_ui_outputs_from_cache(None, None) # Clear previews if new space couldn't be loaded final_overall_status = status_message + f" | Reload Status: {reload_error}" # Keep old UI fields, clear file browser/iframe yield ( gr.update(value=final_overall_status), gr.update(value=_formatted), gr.update(value=_detected), _download, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), [], gr.update(value="*No changes proposed.*"), gr.update(), gr.update(), gr.update(visible=False, choices=[], value=None), gr.update(value=None, visible=False), gr.update(value="*Runtime status unavailable for new space.*") # Clear runtime status ) else: # Should not happen if staged_changeset exists and type is DUPLICATE_SPACE, but as safeguard final_overall_status = "Duplicate Action Error: Invalid duplicate changeset format." # No state change, just report error _formatted, _detected, _download = _generate_ui_outputs_from_cache(owner_name, space_name) yield ( gr.update(value=final_overall_status), gr.update(value=_formatted), gr.update(value=_detected), _download, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), [], gr.update(value="*No changes proposed.*"), gr.update(), gr.update(), gr.update(), gr.update(), gr.update() ) elif is_exclusive_delete: change = first_action # Confirm DELETE_SPACE is only for the currently loaded space before calling the backend delete_owner = change.get('owner') or owner_name delete_space = change.get('space_name') or space_name delete_repo_id_target = f"{delete_owner}/{delete_space}" if delete_owner and delete_space else None current_repo_id = f"{owner_name}/{space_name}" if owner_name and space_name else None if not delete_repo_id_target or delete_repo_id_target != current_repo_id: final_overall_status = f"DELETE_SPACE Error: Action blocked. Cannot delete '{delete_repo_id_target}'. Only deletion of the currently loaded space ('{current_repo_id}') is permitted via AI action." print(f"Blocked DELETE_SPACE action via confirm: requested '{delete_repo_id_target}', current '{current_repo_id}'.") else: status_message = build_logic_delete_space(hf_api_key, delete_owner, delete_space) final_overall_status = status_message # Clear the UI completely after successful deletion if "Successfully" in status_message: parsed_code_blocks_state_cache = [] _formatted, _detected, _download = _generate_ui_outputs_from_cache(None, None) # Clear previews owner_update = gr.update(value="") space_update = gr.update(value="") file_browser_update = gr.update(visible=False, choices=[], value=None) iframe_update = gr.update(value=None, visible=False) runtime_status_update = gr.update(value="*Space deleted, runtime status unavailable.*") else: # If deletion failed, keep the current state and update status _formatted, _detected, _download = _generate_ui_outputs_from_cache(owner_name, space_name) owner_update = gr.update() space_update = gr.update() file_browser_update = gr.update() iframe_update = gr.update() runtime_status_update = gr.update() # Keep existing or trigger refresh? # Let's trigger a status refresh after failed delete attempt runtime_status_update = handle_refresh_space_status(hf_api_key, owner_name, space_name) cleared_changeset = [] yield ( gr.update(value=final_overall_status), gr.update(value=_formatted), gr.update(value=_detected), _download, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), cleared_changeset, gr.update(value="*No changes proposed.*"), owner_update, space_update, file_browser_update, iframe_update, runtime_status_update ) else: # Not an exclusive action, process other actions and file changes sequentially action_status_messages = [] file_change_operations = [] # Process non-file actions first for change in changeset: try: if change['type'] == 'CREATE_SPACE': repo_id_to_create = change.get('repo_id') if repo_id_to_create: msg = build_logic_create_space(hf_api_key, repo_id_to_create, change.get('sdk', 'gradio'), change.get('private', False)) action_status_messages.append(f"CREATE_SPACE: {msg}") else: action_status_messages.append("CREATE_SPACE Error: Target repo_id not specified.") elif change['type'] == 'SET_PRIVACY': repo_id_to_set_privacy = change.get('repo_id') or f"{owner_name}/{space_name}" if repo_id_to_set_privacy and owner_name and space_name: # Ensure current space is loaded msg = build_logic_set_space_privacy(hf_api_key, repo_id_to_set_privacy, change['private']) action_status_messages.append(f"SET_PRIVACY: {msg}") else: action_status_messages.append("SET_PRIVACY Error: Cannot set privacy, no Space currently loaded.") elif change['type'] == 'CREATE_PR': # Ensure source is the currently loaded space source_repo = f"{owner_name}/{space_name}" if owner_name and space_name else None if source_repo and change.get('target_repo_id') and change.get('title'): msg = build_logic_create_pull_request(hf_api_key, source_repo, change['target_repo_id'], change['title'], change.get('body', '')) action_status_messages.append(f"CREATE_PR: {msg}") else: action_status_messages.append("CREATE_PR Error: Source/Target repo, title, or body missing.") elif change['type'] == 'ADD_COMMENT': repo_id_to_comment = change.get('repo_id') or f"{owner_name}/{space_name}" if repo_id_to_comment and owner_name and space_name and change.get('comment'): msg = build_logic_add_comment(hf_api_key, repo_id_to_comment, change['comment']) action_status_messages.append(f"ADD_COMMENT: {msg}") else: action_status_messages.append("ADD_COMMENT Error: Cannot add comment, no Space loaded or comment text missing.") elif change['type'] == 'LIKE_SPACE': repo_id_to_like = change.get('repo_id') or f"{owner_name}/{space_name}" if repo_id_to_like and owner_name and space_name: msg = build_logic_like_space(hf_api_key, repo_id_to_like) action_status_messages.append(f"LIKE_SPACE: {msg}") else: action_status_messages.append("LIKE_SPACE Error: Cannot like space, no Space currently loaded.") elif change['type'] == 'Error': action_status_messages.append(f"Staging Error: {change.get('message', 'Unknown error')}") # Collect file changes for the final commit elif change['type'] in ['CREATE_FILE', 'UPDATE_FILE', 'DELETE_FILE']: file_change_operations.append(change) except Exception as e: action_status_messages.append(f"Error processing action {change.get('type', 'Unknown')}: {e}") print(f"Error processing action {change.get('type', 'Unknown')}: {e}") import traceback traceback.print_exc() # Apply file changes if any were staged file_commit_status = "" if file_change_operations: if owner_name and space_name: file_commit_status = apply_staged_file_changes(hf_api_key, owner_name, space_name, file_change_operations) else: file_commit_status = "File Commit Error: Cannot commit file changes, no Space currently loaded." action_status_messages.append(file_commit_status) # Combine all status messages final_overall_status = " | ".join(action_status_messages) if file_commit_status and file_commit_status != "No file changes (create/update/delete) to commit.": if final_overall_status: final_overall_status += " | " final_overall_status += file_commit_status if not final_overall_status: final_overall_status = "No operations were applied." # After applying changes, reload the space state to reflect the actual state on the Hub _status_reload = f"{final_overall_status} | Reloading Space state..." yield gr.update(value=_status_reload) # Update status refreshed_file_list = [] reload_error = None repo_id_for_reload = f"{owner_name}/{space_name}" if owner_name and space_name else None if repo_id_for_reload: sdk, file_list, err_list = get_space_repository_info(hf_api_key, space_name, owner_name) if err_list: reload_error = f"Error reloading file list after changes: {err_list}" parsed_code_blocks_state_cache = [] else: refreshed_file_list = file_list loaded_files = [] for file_path in refreshed_file_list: content, err_get = get_space_file_content(hf_api_key, space_name, owner_name, file_path) lang = _infer_lang_from_filename(file_path) is_binary = lang == "binary" or (err_get is not None) code = f"[Error loading content: {err_get}]" if err_get else (content or "") loaded_files.append({"filename": file_path, "code": code, "language": lang, "is_binary": is_binary, "is_structure_block": False}) parsed_code_blocks_state_cache = loaded_files file_browser_update = gr.update(visible=True, choices=sorted(refreshed_file_list or []), value=None) if owner_name and space_name: sub_owner = re.sub(r'[^a-z0-9\-]+', '-', owner_name.lower()).strip('-') or 'owner' sub_repo = re.sub(r'[^a-z0-9\-]+', '-', space_name.lower()).strip('-') or 'space' iframe_url = f"https://{sub_owner}-{sub_repo}{'.static.hf.space' if sdk == 'static' else '.hf.space'}" iframe_update = gr.update(value=f'', visible=True) else: iframe_update = gr.update(value=None, visible=False) # Refresh runtime status as well runtime_status_update = handle_refresh_space_status(hf_api_key, owner_name, space_name) else: reload_error = "Cannot reload Space state: Owner or Space Name missing." file_browser_update = gr.update(visible=False, choices=[], value=None) iframe_update = gr.update(value=None, visible=False) runtime_status_update = gr.update(value="*Runtime status unavailable after reload failure.*") _formatted, _detected, _download = _generate_ui_outputs_from_cache(owner_name, space_name) final_overall_status_with_reload = final_overall_status + (f" | Reload Status: {reload_error}" if reload_error else " | Reload Status: Space state refreshed.") cleared_changeset = [] yield ( gr.update(value=final_overall_status_with_reload), gr.update(value=_formatted), gr.update(value=_detected), _download, gr.update(visible=False), gr.update(visible=False), gr.update(visible=False), cleared_changeset, gr.update(value="*No changes proposed.*"), gr.update(), gr.update(), file_browser_update, iframe_update, runtime_status_update ) def handle_cancel_changes(): global parsed_code_blocks_state_cache return ( "Changes cancelled.", [], gr.update(value="*No changes proposed.*"), gr.update(visible=False), gr.update(visible=False), gr.update(visible=False) ) def update_models_dropdown(provider_select): if not provider_select: return gr.update(choices=[], value=None) models = get_models_for_provider(provider_select) default_model = get_default_model_for_provider(provider_select) selected_value = default_model if default_model in models else (models[0] if models else None) return gr.update(choices=models, value=selected_value) def handle_load_existing_space(hf_api_key_ui, ui_owner_name, ui_space_name): global parsed_code_blocks_state_cache _formatted_md_val, _detected_preview_val, _status_val = "*Loading files...*", "*Loading files...*", f"Loading Space: {ui_owner_name}/{ui_space_name}..." _file_browser_update, _iframe_html_update, _download_btn_update = gr.update(visible=False, choices=[], value=None), gr.update(value=None, visible=False), gr.update(interactive=False, value=None) _build_status_clear, _edit_status_clear, _runtime_status_clear = "*Manual build status...*", "*Select a file...*", "*Runtime/Repo status will appear here.*" _changeset_clear = [] _changeset_summary_clear = "*No changes proposed.*" _confirm_ui_hidden = gr.update(visible=False) _list_spaces_display_clear = "*List of spaces will appear here.*" yield ( gr.update(value=_formatted_md_val), gr.update(value=_detected_preview_val), gr.update(value=_status_val), _file_browser_update, gr.update(value=ui_owner_name), gr.update(value=ui_space_name), _iframe_html_update, _download_btn_update, gr.update(value=_build_status_clear), gr.update(value=_edit_status_clear), gr.update(value=_runtime_status_clear), _changeset_clear, gr.update(value=_changeset_summary_clear), _confirm_ui_hidden, _confirm_ui_hidden, _confirm_ui_hidden, gr.update(value=_list_spaces_display_clear) ) owner_to_use = ui_owner_name token, token_err = build_logic_get_api_token(hf_api_key_ui) if token_err: _status_val = f"Load Error: {token_err}" yield gr.update(value=_status_val), return if not owner_to_use: try: user_info = build_logic_whoami(token=token) owner_to_use = user_info.get('name') if not owner_to_use: raise Exception("Could not find user name from token.") yield gr.update(value=owner_to_use), gr.update(value=f"Loading Space: {owner_to_use}/{ui_space_name} (Auto-detected owner)...") except Exception as e: _status_val = f"Load Error: Error auto-detecting owner: {e}" yield gr.update(value=_status_val), return if not owner_to_use or not ui_space_name: _status_val = "Load Error: Owner and Space Name are required." yield gr.update(value=_status_val), return sdk, file_list, err = get_space_repository_info(hf_api_key_ui, ui_space_name, owner_to_use) yield gr.update(value=owner_to_use), gr.update(value=ui_space_name), if err: _status_val = f"Load Error: {err}" parsed_code_blocks_state_cache = [] _formatted, _detected, _download = _generate_ui_outputs_from_cache(owner_to_use, ui_space_name) yield ( gr.update(value=_formatted), gr.update(value=_detected), gr.update(value=_status_val), gr.update(visible=False, choices=[], value=None), gr.update(), gr.update(), gr.update(value=None, visible=False), _download, gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update() ) return loaded_files = [] for file_path in file_list: content, err_get = get_space_file_content(hf_api_key_ui, ui_space_name, owner_to_use, file_path) lang = _infer_lang_from_filename(file_path) is_binary = lang == "binary" or (err_get is not None) code = f"[Error loading content: {err_get}]" if err_get else (content or "") loaded_files.append({"filename": file_path, "code": code, "language": lang, "is_binary": is_binary, "is_structure_block": False}) parsed_code_blocks_state_cache = loaded_files _formatted, _detected, _download = _generate_ui_outputs_from_cache(owner_to_use, ui_space_name) _status_val = f"Successfully loaded {len(file_list)} files from {owner_to_use}/{ui_space_name}. SDK: {sdk or 'unknown'}." file_browser_update = gr.update(visible=True, choices=sorted(file_list or []), value=None) if owner_to_use and ui_space_name: sub_owner = re.sub(r'[^a-z0-9\-]+', '-', owner_to_use.lower()).strip('-') or 'owner' sub_repo = re.sub(r'[^a-z0-9\-]+', '-', ui_space_name.lower()).strip('-') or 'space' iframe_url = f"https://{sub_owner}-{sub_repo}{'.static.hf.space' if sdk == 'static' else '.hf.space'}" iframe_update = gr.update(value=f'', visible=True) else: iframe_update = gr.update(value=None, visible=False) runtime_status_md = handle_refresh_space_status(hf_api_key_ui, owner_to_use, ui_space_name) yield ( gr.update(value=_formatted), gr.update(value=_detected), gr.update(value=_status_val), file_browser_update, gr.update(), gr.update(), iframe_update, _download, gr.update(), gr.update(), gr.update(value=runtime_status_md), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update() ) def handle_build_space_button(hf_api_key_ui, ui_space_name_part, ui_owner_name_part, space_sdk_ui, is_private_ui, formatted_markdown_content): global parsed_code_blocks_state_cache _build_status = "Starting manual space build process..." _iframe_html, _file_browser_update = gr.update(value=None, visible=False), gr.update(visible=False, choices=[], value=None) _changeset_clear = [] _changeset_summary_clear = "*Manual build initiated, changes plan cleared.*" _confirm_ui_hidden = gr.update(visible=False) yield (gr.update(value=_build_status), _iframe_html, _file_browser_update, gr.update(value=ui_owner_name_part), gr.update(value=ui_space_name_part), _changeset_clear, gr.update(value=_changeset_summary_clear), _confirm_ui_hidden, _confirm_ui_hidden, _confirm_ui_hidden, gr.update(), gr.update(), gr.update(), gr.update()) if not ui_space_name_part or "/" in ui_space_name_part: _build_status = f"Build Error: Invalid Space Name '{ui_space_name_part}'." yield gr.update(value=_build_status), return parsed_content = build_logic_parse_markdown(formatted_markdown_content) proposed_files_list = parsed_content.get("files", []) manual_changeset = [] if ui_owner_name_part and ui_space_name_part: manual_changeset.append({"type": "CREATE_SPACE", "repo_id": f"{ui_owner_name_part}/{ui_space_name_part}", "sdk": space_sdk_ui, "private": is_private_ui}) for file_info in proposed_files_list: manual_changeset.append({"type": "CREATE_FILE", "path": file_info["path"], "content": file_info["content"], "lang": _infer_lang_from_filename(file_info["path"])}) if not manual_changeset: _build_status = "Build Error: No target space specified or no files parsed from markdown." yield gr.update(value=_build_status), return # Use apply_staged_file_changes and CREATE_SPACE action via confirm handler logic # Or simulate the parts needed here directly? Direct call might be simpler for manual build. # Let's use apply_staged_file_changes directly, but need to handle the CREATE_SPACE part separately first. token, token_err = build_logic_get_api_token(hf_api_key_ui) if token_err: _build_status = f"Build Error: API Token Error: {token_err}" yield gr.update(value=_build_status), return api = HfApi(token=token) repo_id_target = f"{ui_owner_name_part}/{ui_space_name_part}" build_status_messages = [] try: # Attempt to create/ensure space exists first api.create_repo(repo_id=repo_id_target, repo_type="space", space_sdk=space_sdk_select, private=is_private_ui, exist_ok=True) build_status_messages.append(f"CREATE_SPACE: Successfully created or ensured space [{repo_id_target}](https://huggingface.co/spaces/{repo_id_target}) exists.") except HfHubHTTPError as e_http: build_status_messages.append(f"CREATE_SPACE HTTP Error ({e_http.response.status_code if e_http.response else 'N/A'}): {e_http.response.text if e_http.response else str(e_http)}. Check logs.") if e_http.response and e_http.response.status_code in (401, 403): _build_status = f"Build Error: {build_status_messages[-1]}. Cannot proceed with file upload."; yield gr.update(value=_build_status); return except Exception as e: build_status_messages.append(f"CREATE_SPACE Error: {e}") _build_status = f"Build Error: {build_status_messages[-1]}. Cannot proceed with file upload."; yield gr.update(value=_build_status); return # Now apply file changes from the markdown content file_changes_only_changeset = [{"type": "CREATE_FILE", "path": f["path"], "content": f["content"], "lang": _infer_lang_from_filename(f["path"])} for f in proposed_files_list] if file_changes_only_changeset: file_commit_status = apply_staged_file_changes(hf_api_key_ui, ui_owner_name_part, ui_space_name_part, file_changes_only_changeset) build_status_messages.append(file_commit_status) else: build_status_messages.append("No files parsed from markdown to upload/update.") _build_status = " | ".join(build_status_messages) owner_to_use = ui_owner_name_part space_to_use = ui_space_name_part _formatted_md = formatted_markdown_content _detected_preview_val = "*Loading files after build...*" _download_btn_update = gr.update(interactive=False, value=None) yield ( gr.update(value=_build_status), _iframe_html, _file_browser_update, gr.update(value=owner_to_use), gr.update(value=space_to_use), _changeset_clear, gr.update(value=_changeset_summary_clear), _confirm_ui_hidden, _confirm_ui_hidden, _confirm_ui_hidden, gr.update(value=_formatted_md), gr.update(value=_detected_preview_val), _download_btn_update, gr.update() # Keep runtime status ) sdk_built, file_list, err_list = get_space_repository_info(hf_api_key_ui, space_to_use, owner_to_use) if err_list: _build_status += f" | Error reloading file list after build: {err_list}" parsed_code_blocks_state_cache = [] _file_browser_update = gr.update(visible=False, choices=[], value=None) _iframe_html_update = gr.update(value=None, visible=False) else: loaded_files = [] for file_path in file_list: content, err_get = get_space_file_content(hf_api_key_ui, space_to_use, owner_to_use, file_path) lang = _infer_lang_from_filename(file_path) is_binary = lang == "binary" or (err_get is not None) code = f"[Error loading content: {err_get}]" if err_get else (content or "") loaded_files.append({"filename": file_path, "code": code, "language": lang, "is_binary": is_binary, "is_structure_block": False}) parsed_code_blocks_state_cache = loaded_files _file_browser_update = gr.update(visible=True, choices=sorted(file_list or []), value=None) if owner_to_use and space_to_use: sub_owner = re.sub(r'[^a-z0-9\-]+', '-', owner_to_use.lower()).strip('-') or 'owner' sub_repo = re.sub(r'[^a-z0-9\-]+', '-', space_to_use.lower()).strip('-') or 'space' iframe_url = f"https://{sub_owner}-{sub_repo}{'.static.hf.space' if sdk_built == 'static' else '.hf.space'}" _iframe_html_update = gr.update(value=f'', visible=True) else: _iframe_html_update = gr.update(value=None, visible=False) runtime_status_md = handle_refresh_space_status(hf_api_key_ui, owner_to_use, space_to_use) yield ( gr.update(value=_build_status), _iframe_html_update, _file_browser_update, gr.update(value=owner_to_use), gr.update(value=space_to_use), _changeset_clear, gr.update(value=_changeset_summary_clear), _confirm_ui_hidden, _confirm_ui_hidden, _confirm_ui_hidden, gr.update(value=_formatted_md), gr.update(value=_detected_preview), _download, gr.update(value=runtime_status_md) ) def handle_load_file_for_editing(hf_api_key_ui, ui_space_name_part, ui_owner_name_part, selected_file_path): if not selected_file_path: return "", "Select a file.", "", gr.update(language="plaintext") content, err = get_space_file_content(hf_api_key_ui, ui_space_name_part, ui_owner_name_part, selected_file_path) if err: return "", f"Load Error: {err}", "", gr.update(language="plaintext") lang = _infer_lang_from_filename(selected_file_path) commit_msg = f"Update {selected_file_path}" return content, f"Loaded `{selected_file_path}`", commit_msg, gr.update(language=lang) def handle_commit_file_changes(hf_api_key_ui, ui_space_name_part, ui_owner_name_part, file_to_edit_path, edited_content, commit_message): if not ui_owner_name_part or not ui_space_name_part: return "Commit Error: Cannot commit changes. Please load a space first.", gr.update(), gr.update(), gr.update(), gr.update() if not file_to_edit_path: return "Commit Error: No file selected for commit.", gr.update(), gr.update(), gr.update(), gr.update() status_msg = update_space_file(hf_api_key_ui, ui_space_name_part, ui_owner_name_part, file_to_edit_path, edited_content, commit_message) global parsed_code_blocks_state_cache file_browser_update = gr.update() if "Successfully" in status_msg: found = False for block in parsed_code_blocks_state_cache: if block["filename"] == file_to_edit_path and not block.get("is_structure_block"): block["code"] = edited_content block["language"] = _infer_lang_from_filename(file_to_edit_path) block["is_binary"] = False found = True break if not found: parsed_code_blocks_state_cache.append({ "filename": file_to_edit_path, "code": edited_content, "language": _infer_lang_from_filename(file_to_edit_path), "is_binary": False, "is_structure_block": False }) parsed_code_blocks_state_cache.sort(key=lambda b: (0, b["filename"]) if b.get("is_structure_block") else (1, b["filename"])) file_list, _ = list_space_files_for_browsing(hf_api_key_ui, ui_space_name_part, ui_owner_name_part) file_browser_update = gr.update(choices=sorted(file_list or [])) _formatted, _detected, _download = _generate_ui_outputs_from_cache(ui_owner_name_part, ui_space_name_part) return status_msg, file_browser_update, gr.update(value=_formatted), gr.update(value=_detected), _download def handle_delete_file(hf_api_key_ui, ui_space_name_part, ui_owner_name_part, file_to_delete_path): if not ui_owner_name_part or not ui_space_name_part: return "Delete Error: Cannot delete file. Please load a space first.", gr.update(), "", "", "plaintext", gr.update(), gr.update(), gr.update() if not file_to_delete_path: return "Delete Error: No file selected to delete.", gr.update(), "", "", "plaintext", gr.update(), gr.update(), gr.update() status_msg = build_logic_delete_space_file(hf_api_key_ui, ui_space_name_part, ui_owner_name_part, file_to_delete_path) global parsed_code_blocks_state_cache file_browser_update = gr.update() file_content_editor_update = gr.update() commit_message_update = gr.update() editor_lang_update = gr.update() if "Successfully" in status_msg: parsed_code_blocks_state_cache = [b for b in parsed_code_blocks_state_cache if b["filename"] != file_to_delete_path] file_content_editor_update = gr.update(value="") commit_message_update = gr.update(value="") editor_lang_update = gr.update(language="plaintext") file_list, _ = list_space_files_for_browsing(hf_api_key_ui, ui_space_name_part, ui_owner_name_part) file_browser_update = gr.update(choices=sorted(file_list or []), value=None) _formatted, _detected, _download = _generate_ui_outputs_from_cache(ui_owner_name_part, ui_space_name_part) return ( status_msg, file_browser_update, file_content_editor_update, commit_message_update, editor_lang_update, gr.update(value=_formatted), gr.update(value=_detected), _download ) def handle_refresh_space_status(hf_api_key_ui, ui_owner_name, ui_space_name): if not ui_owner_name or not ui_space_name: return "**Status Error:** Owner and Space Name must be provided to get status." status_details, err = get_space_runtime_status(hf_api_key_ui, ui_space_name, ui_owner_name) repo_info = None repo_info_err = None if not err: # Only fetch repo info if runtime status succeeded (implies space exists) sdk, file_list, repo_info_err = get_space_repository_info(hf_api_key_ui, ui_space_name, ui_owner_name) if not repo_info_err: repo_info = {"sdk": sdk, "file_count": len(file_list) if file_list is not None else 'N/A'} md_lines = [] md_lines.append(f"### Status for {ui_owner_name}/{ui_space_name}\n") if err: md_lines.append(f"- **Runtime Status Error:** {err}\n") elif not status_details: md_lines.append("*Could not retrieve runtime status details.*\n") else: md_lines.append(f"- **Stage:** `{status_details.get('stage', 'N/A')}`\n") md_lines.append(f"- **Status:** `{status_details.get('status', 'N/A')}`\n") md_lines.append(f"- **Hardware:** `{status_details.get('hardware', 'N/A')}`\n") requested_hw = status_details.get('requested_hardware') if requested_hw: md_lines.append(f"- **Requested Hardware:** `{requested_hw}`\n") error_msg = status_details.get('error_message') if error_msg: md_lines.append(f"- **Error:** `{escape_html_for_markdown(error_msg)}`\n") log_link = status_details.get('full_log_link') if log_link and log_link != "#": md_lines.append(f"- [View Full Logs]({log_link})\n") md_lines.append("---") md_lines.append("### Repository Info\n") if repo_info_err: md_lines.append(f"- **Repo Info Error:** {repo_info_err}\n") elif repo_info: md_lines.append(f"- **SDK:** `{repo_info.get('sdk', 'N/A')}`\n") md_lines.append(f"- **File Count:** `{repo_info.get('file_count', 'N/A')}`\n") else: md_lines.append("*Could not retrieve repository info.*") return "\n".join(md_lines) def handle_list_spaces(hf_api_key_ui, ui_owner_name): token, token_err = build_logic_get_api_token(hf_api_key_ui) if token_err: return f"**List Spaces Error:** {token_err}" owner_to_list = ui_owner_name if not owner_to_list: try: user_info = build_logic_whoami(token=token) owner_to_list = user_info.get('name') if not owner_to_list: raise Exception("Could not find user name from token.") except Exception as e: return f"**List Spaces Error:** Error auto-detecting owner: {e}. Please specify Owner field." if not owner_to_list: return "**List Spaces Error:** Owner could not be determined. Please specify it in the Owner field." spaces_list, err = build_logic_list_user_spaces(hf_api_key=token, owner=owner_to_list) if err: return f"**List Spaces Error:** {err}" if not spaces_list: return f"*No spaces found for user/org `{owner_to_list}`.*" md = f"### Spaces for `{owner_to_list}`\n" for space in sorted(spaces_list): md += f"- `{space}`\n" return md def handle_manual_duplicate_space(hf_api_key_ui, source_owner, source_space_name, target_owner, target_space_name, target_private): if not source_owner or not source_space_name: return "Duplicate Error: Please load a Space first to duplicate.", gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update() if not target_owner or not target_space_name: return "Duplicate Error: Target Owner and Target Space Name are required.", gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update() if "/" in target_space_name: return "Duplicate Error: Target Space Name should not contain '/'. Use Target Owner field for the owner part.", gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update() global parsed_code_blocks_state_cache source_repo_id = f"{source_owner}/{source_space_name}" target_repo_id = f"{target_owner}/{target_space_name}" status_msg = f"Attempting to duplicate `{source_repo_id}` to `{target_repo_id}`..." # Yield initial status yield status_msg, gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update() result_message = build_logic_duplicate_space(hf_api_key_ui, source_repo_id, target_repo_id, target_private) status_msg = f"Duplication Result: {result_message}" _status_reload = f"{status_msg} | Attempting to load the new Space [{target_repo_id}]..." yield gr.update(value=_status_reload) # Update status new_owner = target_owner new_space_name = target_space_name sdk, file_list, err_list = get_space_repository_info(hf_api_key_ui, new_space_name, new_owner) if err_list: reload_error = f"Error reloading file list after duplication: {err_list}" parsed_code_blocks_state_cache = [] _file_browser_update = gr.update(visible=False, choices=[], value=None) _iframe_html_update = gr.update(value=None, visible=False) else: loaded_files = [] for file_path in file_list: content, err_get = get_space_file_content(hf_api_key_ui, new_space_name, new_owner, file_path) lang = _infer_lang_from_filename(file_path) is_binary = lang == "binary" or (err_get is not None) code = f"[Error loading content: {err_get}]" if err_get else (content or "") loaded_files.append({"filename": file_path, "code": code, "language": lang, "is_binary": is_binary, "is_structure_block": False}) parsed_code_blocks_state_cache = loaded_files _file_browser_update = gr.update(visible=True, choices=sorted([f["filename"] for f in parsed_code_blocks_state_cache if not f.get("is_structure_block")] or []), value=None) if new_owner and new_space_name: sub_owner = re.sub(r'[^a-z0-9\-]+', '-', new_owner.lower()).strip('-') or 'owner' sub_repo = re.sub(r'[^a-z0-9\-]+', '-', new_space_name.lower()).strip('-') or 'space' iframe_url = f"https://{sub_owner}-{sub_repo}{'.static.hf.space' if sdk == 'static' else '.hf.space'}" _iframe_html_update = gr.update(value=f'', visible=True) else: _iframe_html_update = gr.update(value=None, visible=False) _formatted, _detected, _download = _generate_ui_outputs_from_cache(new_owner, new_space_name) final_overall_status = status_msg + (f" | Reload Status: {reload_error}" if reload_error else f" | Reload Status: New Space [{new_owner}/{new_space_name}] state refreshed.") runtime_status_md = handle_refresh_space_status(hf_api_key_ui, new_owner, new_space_name) owner_update = gr.update(value=new_owner) space_update = gr.update(value=new_space_name) yield ( gr.update(value=final_overall_status), owner_update, space_update, gr.update(value=_formatted), gr.update(value=_detected), _download, _file_browser_update, _iframe_html_update, gr.update(value=runtime_status_md) ) custom_theme = gr.themes.Base(primary_hue="teal", secondary_hue="purple", neutral_hue="zinc", text_size="sm", spacing_size="md", radius_size="sm", font=["System UI", "sans-serif"]) custom_css = """ body { background: linear-gradient(to bottom right, #2c3e50, #34495e); color: #ecf0f1; } .gradio-container { background: transparent !important; } .gr-box, .gr-panel, .gr-pill { background-color: rgba(44, 62, 80, 0.8) !important; border-color: rgba(189, 195, 199, 0.2) !important; } .gr-textbox, .gr-dropdown, .gr-button, .gr-code, .gr-chat-message { border-color: rgba(189, 195, 199, 0.3) !important; background-color: rgba(52, 73, 94, 0.9) !important; color: #ecf0f1 !important; } .gr-button.gr-button-primary { background-color: #1abc9c !important; color: white !important; border-color: #16a085 !important; } .gr-button.gr-button-secondary { background-color: #9b59b6 !important; color: white !important; border-color: #8e44ad !important; } .gr-button.gr-button-stop { background-color: #e74c3c !important; color: white !important; border-color: #c0392b !important; } .gr-markdown { background-color: rgba(44, 62, 80, 0.7) !important; padding: 10px; border-radius: 5px; overflow-x: auto; } .gr-markdown h1, .gr-markdown h2, .gr-markdown h3, .gr-markdown h4, .gr-markdown h5, .gr-markdown h6 { color: #ecf0f1 !important; border-bottom-color: rgba(189, 195, 199, 0.3) !important; } .gr-markdown pre code { background-color: rgba(52, 73, 94, 0.95) !important; border-color: rgba(189, 195, 199, 0.3) !important; } .gr-chatbot { background-color: rgba(44, 62, 80, 0.7) !important; border-color: rgba(189, 195, 199, 0.2) !important; } .gr-chatbot .message { background-color: rgba(52, 73, 94, 0.9) !important; color: #ecf0f1 !important; border-color: rgba(189, 195, 199, 0.3) !important; } .gr-chatbot .message.user { background-color: rgba(46, 204, 113, 0.9) !important; color: black !important; } .gradio-container .gr-accordion { border-color: rgba(189, 195, 199, 0.3) !important; } .gradio-container .gr-accordion.closed { background-color: rgba(52, 73, 94, 0.9) !important; } .gradio-container .gr-accordion.open { background-color: rgba(44, 62, 80, 0.8) !important; } """ with gr.Blocks(theme=custom_theme, css=custom_css) as demo: changeset_state = gr.State([]) gr.Markdown("# 🤖 AI-Powered Hugging Face Space Commander") gr.Markdown("Use an AI assistant to create, modify, build, and manage your Hugging Face Spaces directly from this interface.") gr.Markdown("## ❗ This will cause changes to your huggingface spaces if you give it your Huggingface Key") gr.Markdown("⚒ Under Development - Use with Caution") with gr.Sidebar(): with gr.Column(scale=1): with gr.Accordion("⚙️ Configuration", open=True): hf_api_key_input = gr.Textbox(label="Hugging Face Token", type="password", placeholder="hf_... (uses env var HF_TOKEN if empty)") owner_name_input = gr.Textbox(label="HF Owner Name", placeholder="e.g., your-username") space_name_input = gr.Textbox(label="HF Space Name", value="") load_space_button = gr.Button("🔄 Load Existing Space", variant="secondary") with gr.Accordion("🤖 AI Model Settings", open=True): available_providers = get_available_providers() default_provider = 'Groq' if default_provider not in available_providers: default_provider = available_providers[0] if available_providers else None elif len(available_providers) < 3: default_provider = available_providers[0] if available_providers else None initial_models = get_models_for_provider(default_provider) if default_provider else [] initial_model = get_default_model_for_provider(default_provider) if default_provider else None if initial_model not in initial_models: initial_model = initial_models[0] if initial_models else None provider_select = gr.Dropdown( label="AI Provider", choices=available_providers, value=default_provider, allow_custom_value=False ) model_select = gr.Dropdown( label="AI Model", choices=initial_models, value=initial_model, allow_custom_value=False ) provider_api_key_input = gr.Textbox(label="Model Provider API Key (Optional)", type="password", placeholder="sk_... (overrides backend settings)") system_prompt_input = gr.Textbox(label="System Prompt", lines=10, value=DEFAULT_SYSTEM_PROMPT, elem_id="system-prompt") with gr.Accordion("🗄️ Space Management", open=True): gr.Markdown("### Manual Actions") with gr.Group(): gr.Markdown("Duplicate Current Space To:") target_owner_input = gr.Textbox(label="Target Owner Name", placeholder="e.g., new-username") target_space_name_input = gr.Textbox(label="Target Space Name", placeholder="e.g., new-space") target_private_checkbox = gr.Checkbox(label="Make Target Private", value=False) duplicate_space_button = gr.Button("📂 Duplicate Space", variant="secondary") gr.Markdown("---") gr.Markdown("### List Spaces") list_spaces_button = gr.Button("📄 List My Spaces", variant="secondary") list_spaces_display = gr.Markdown("*List of spaces will appear here.*") with gr.Column(scale=2): gr.Markdown("## 💬 AI Assistant Chat") chatbot_display = gr.Chatbot(label="AI Chat", height=500, bubble_full_width=False, avatar_images=(None)) with gr.Row(): chat_message_input = gr.Textbox(show_label=False, placeholder="Your Message...", scale=7) send_chat_button = gr.Button("Send", variant="primary", scale=1) status_output = gr.Textbox(label="Last Action Status", interactive=False, value="Ready.") with gr.Accordion("📝 Proposed Changes (Pending Confirmation)", open=False, visible=False) as confirm_accordion: changeset_display = gr.Markdown("*No changes proposed.*") with gr.Row(): confirm_button = gr.Button("✅ Confirm & Apply Changes", variant="primary", visible=False) cancel_button = gr.Button("❌ Cancel", variant="stop", visible=False) with gr.Tabs(): with gr.TabItem("📝 Generated Markdown & Build"): with gr.Row(): with gr.Column(scale=2): formatted_space_output_display = gr.Textbox(label="Current Space Definition (Generated Markdown)", lines=20, interactive=True, value="*Load or create a space to see its definition.*") download_button = gr.DownloadButton(label="Download .md", interactive=False) with gr.Column(scale=1): gr.Markdown("### Manual Build & Status") space_sdk_select = gr.Dropdown(label="Space SDK", choices=["gradio", "streamlit", "docker", "static"], value="gradio", interactive=True) space_private_checkbox = gr.Checkbox(label="Make Space Private", value=False, interactive=True) build_space_button = gr.Button("🚀 Build / Update Space from Markdown", variant="primary") build_status_display = gr.Textbox(label="Manual Build/Update Status", interactive=False, value="*Manual build status...*") gr.Markdown("---") refresh_status_button = gr.Button("🔄 Refresh Runtime/Repo Status") space_runtime_status_display = gr.Markdown("*Runtime/Repo status will appear here.*") with gr.TabItem("🔍 Files Preview"): detected_files_preview = gr.Markdown(value="*A preview of the latest file versions will appear here.*") with gr.TabItem("✏️ Live File Editor & Preview"): with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Live Editor") file_browser_dropdown = gr.Dropdown(label="Select File in Space", choices=[], interactive=True) file_content_editor = gr.Code(label="File Content Editor", language="python", lines=15, interactive=True, value="") commit_message_input = gr.Textbox(label="Commit Message", placeholder="e.g., Updated app.py", interactive=True, value="") with gr.Row(): update_file_button = gr.Button("Commit Changes", variant="primary", interactive=True) delete_file_button = gr.Button("🗑️ Delete Selected File", variant="stop", interactive=True) edit_status_display = gr.Textbox(label="File Edit/Delete Status", interactive=False, value="") with gr.Column(scale=1): gr.Markdown("### Live Space Preview") space_iframe_display = gr.HTML(value="", visible=True) provider_select.change(update_models_dropdown, inputs=provider_select, outputs=model_select) chat_inputs = [ chat_message_input, chatbot_display, hf_api_key_input, provider_api_key_input, provider_select, model_select, system_prompt_input, owner_name_input, space_name_input ] chat_outputs = [ chat_message_input, chatbot_display, status_output, detected_files_preview, formatted_space_output_display, download_button, changeset_state, changeset_display, confirm_accordion, confirm_button, cancel_button ] send_chat_button.click(handle_chat_submit, inputs=chat_inputs, outputs=chat_outputs) chat_message_input.submit(handle_chat_submit, inputs=chat_inputs, outputs=chat_outputs) confirm_inputs = [hf_api_key_input, owner_name_input, space_name_input, changeset_state] confirm_outputs = [ status_output, formatted_space_output_display, detected_files_preview, download_button, confirm_accordion, confirm_button, cancel_button, changeset_state, changeset_display, owner_name_input, space_name_input, file_browser_dropdown, space_iframe_display, space_runtime_status_display # Added outputs for potential state changes ] confirm_button.click(handle_confirm_changes, inputs=confirm_inputs, outputs=confirm_outputs) cancel_outputs = [ status_output, changeset_state, changeset_display, confirm_accordion, confirm_button, cancel_button ] cancel_button.click(handle_cancel_changes, inputs=None, outputs=cancel_outputs) load_space_outputs = [ formatted_space_output_display, detected_files_preview, status_output, file_browser_dropdown, owner_name_input, space_name_input, space_iframe_display, download_button, build_status_display, edit_status_display, space_runtime_status_display, changeset_state, changeset_display, confirm_accordion, confirm_button, cancel_button, list_spaces_display # Added outputs for List Spaces ] load_space_button.click( fn=handle_load_existing_space, inputs=[hf_api_key_input, owner_name_input, space_name_input], outputs=load_space_outputs ) build_outputs = [ build_status_display, space_iframe_display, file_browser_dropdown, owner_name_input, space_name_input, changeset_state, changeset_display, confirm_accordion, confirm_button, cancel_button, formatted_space_output_display, detected_files_preview, download_button, space_runtime_status_display ] build_inputs = [ hf_api_key_input, space_name_input, owner_name_input, space_sdk_select, space_private_checkbox, formatted_space_output_display ] build_space_button.click(fn=handle_build_space_button, inputs=build_inputs, outputs=build_outputs) file_edit_load_outputs = [file_content_editor, edit_status_display, commit_message_input, file_content_editor] file_browser_dropdown.change(fn=handle_load_file_for_editing, inputs=[hf_api_key_input, space_name_input, owner_name_input, file_browser_dropdown], outputs=file_edit_load_outputs) commit_file_outputs = [edit_status_display, file_browser_dropdown, formatted_space_output_display, detected_files_preview, download_button] update_file_button.click(fn=handle_commit_file_changes, inputs=[hf_api_key_input, space_name_input, owner_name_input, file_browser_dropdown, file_content_editor, commit_message_input], outputs=commit_file_outputs) delete_file_outputs = [ edit_status_display, file_browser_dropdown, file_content_editor, commit_message_input, file_content_editor, formatted_space_output_display, detected_files_preview, download_button ] delete_file_button.click(fn=handle_delete_file, inputs=[hf_api_key_input, space_name_input, owner_name_input, file_browser_dropdown], outputs=delete_file_outputs) refresh_status_button.click(fn=handle_refresh_space_status, inputs=[hf_api_key_input, owner_name_input, space_name_input], outputs=[space_runtime_status_display]) manual_duplicate_inputs = [hf_api_key_input, owner_name_input, space_name_input, target_owner_input, target_space_name_input, target_private_checkbox] manual_duplicate_outputs = [ status_output, owner_name_input, space_name_input, formatted_space_output_display, detected_files_preview, download_button, file_browser_dropdown, space_iframe_display, space_runtime_status_display ] duplicate_space_button.click(fn=handle_manual_duplicate_space, inputs=manual_duplicate_inputs, outputs=manual_duplicate_outputs) list_spaces_button.click(fn=handle_list_spaces, inputs=[hf_api_key_input, owner_name_input], outputs=[list_spaces_display]) if __name__ == "__main__": demo.launch(debug=False, mcp_server=True)