Spaces:

launch-calcium
/

Qwen-Code3

Build error

File size: 13,188 Bytes

a1e0cbd

import os
import re
import shutil
import subprocess
import tempfile
import gradio as gr

# Ensure workspace folder exists
WORKSPACE_DIR = os.path.abspath("./workspace")
os.makedirs(WORKSPACE_DIR, exist_ok=True)

# Helper to strip ANSI escape codes from terminal outputs
def strip_ansi(text):
    ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
    return ansi_escape.sub('', text)

# Helper to get files list
def list_workspace_files():
    files_list = []
    for root, dirs, files in os.walk(WORKSPACE_DIR):
        # Ignore hidden files/directories like .git or .qwen
        if ".git" in root or ".qwen" in root:
            continue
        for file in files:
            full_path = os.path.join(root, file)
            rel_path = os.path.relpath(full_path, WORKSPACE_DIR)
            files_list.append(rel_path)
    return sorted(files_list)

# Load file content
def load_file_content(filepath):
    if not filepath:
        return "No file selected."
    full_path = os.path.join(WORKSPACE_DIR, filepath)
    if not os.path.exists(full_path) or os.path.isdir(full_path):
        return f"File {filepath} not found or is a directory."
    try:
        with open(full_path, "r", encoding="utf-8", errors="ignore") as f:
            return f.read()
    except Exception as e:
        return f"Error reading file: {str(e)}"

# Save file content
def save_file_content(filepath, content):
    if not filepath:
        return "No file selected.", list_workspace_files()
    full_path = os.path.join(WORKSPACE_DIR, filepath)
    try:
        os.makedirs(os.path.dirname(full_path), exist_ok=True)
        with open(full_path, "w", encoding="utf-8") as f:
            f.write(content)
        return f"Successfully saved {filepath}!", gr.update(choices=list_workspace_files(), value=filepath)
    except Exception as e:
        return f"Error saving file: {str(e)}", list_workspace_files()

# Initialize templates
def load_template(template_name):
    # Clear existing workspace
    for item in os.listdir(WORKSPACE_DIR):
        item_path = os.path.join(WORKSPACE_DIR, item)
        if item == ".qwen":
            continue
        if os.path.isdir(item_path):
            shutil.rmtree(item_path)
        else:
            os.remove(item_path)

    if template_name == "Python Math Library":
        os.makedirs(os.path.join(WORKSPACE_DIR, "pymath"), exist_ok=True)
        with open(os.path.join(WORKSPACE_DIR, "pymath", "math.py"), "w") as f:
            f.write("def add(a, b):\n    return a + b\n\ndef multiply(a, b):\n    return a * b\n")
        with open(os.path.join(WORKSPACE_DIR, "README.md"), "w") as f:
            f.write("# Python Math Library\n\nThis is a simple math library. Ask Qwen-Code to add features or tests!\n")
    elif template_name == "NodeJS Web App":
        with open(os.path.join(WORKSPACE_DIR, "package.json"), "w") as f:
            f.write('{\n  "name": "simple-web-app",\n  "version": "1.0.0",\n  "main": "index.js",\n  "dependencies": {}\n}\n')
        with open(os.path.join(WORKSPACE_DIR, "index.js"), "w") as f:
            f.write("console.log('Hello from NodeJS web app!');\n")
    elif template_name == "HTML Landing Page":
        with open(os.path.join(WORKSPACE_DIR, "index.html"), "w") as f:
            f.write("<!DOCTYPE html>\n<html>\n<head>\n  <title>My Landing Page</title>\n</head>\n<body>\n  <h1>Welcome to my Qwen-Code Generated Page!</h1>\n</body>\n</html>\n")
    
    return f"Initialized workspace with '{template_name}' template.", gr.update(choices=list_workspace_files())

# Zip workspace for download
def download_workspace():
    temp_dir = tempfile.gettempdir()
    zip_path = os.path.join(temp_dir, "workspace_archive")
    shutil.make_archive(zip_path, 'zip', WORKSPACE_DIR)
    return zip_path + ".zip"

# Execute Qwen-Code agent
def run_agent(provider, api_key, model_name, custom_model, prompt, history):
    if not api_key:
        yield history, "⚠️ Please enter an API key under API settings first!", gr.update()
        return

    selected_model = custom_model if model_name == "custom" else model_name
    if not selected_model:
        yield history, "⚠️ Please select or specify a model first!", gr.update()
        return

    # Prepare command and env
    cmd = ["qwen", "-p", prompt, "--yolo"]
    env = os.environ.copy()
    env["QWEN_SANDBOX"] = "false"  # Run inside the container's environment natively

    # Configure env variables based on provider
    if provider == "DashScope (Qwen-compatible)":
        env["DASHSCOPE_API_KEY"] = api_key
        env["OPENAI_API_KEY"] = api_key
        env["OPENAI_BASE_URL"] = "https://dashscope.aliyuncs.com/compatible-mode/v1"
        env["OPENAI_MODEL"] = selected_model
    elif provider == "Alibaba Cloud Coding Plan (Beijing)":
        env["BAILIAN_CODING_PLAN_API_KEY"] = api_key
        env["OPENAI_BASE_URL"] = "https://coding.dashscope.aliyuncs.com/v1"
        env["OPENAI_MODEL"] = selected_model
    elif provider == "Alibaba Cloud Coding Plan (International)":
        env["BAILIAN_CODING_PLAN_API_KEY"] = api_key
        env["OPENAI_BASE_URL"] = "https://coding-intl.dashscope.aliyuncs.com/v1"
        env["OPENAI_MODEL"] = selected_model
    elif provider == "OpenAI":
        env["OPENAI_API_KEY"] = api_key
        env["OPENAI_MODEL"] = selected_model
    elif provider == "Anthropic Claude":
        env["ANTHROPIC_API_KEY"] = api_key
        env["ANTHROPIC_MODEL"] = selected_model
    elif provider == "Google Gemini":
        env["GEMINI_API_KEY"] = api_key
        env["GEMINI_MODEL"] = selected_model

    history.append((prompt, "... Starting Qwen Code Agent ..."))
    yield history, "Initializing Qwen Code CLI process...", gr.update()

    # Launch subprocess and read stdout in real-time
    console_output = ""
    try:
        proc = subprocess.Popen(
            cmd,
            cwd=WORKSPACE_DIR,
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1
        )

        for line in iter(proc.stdout.readline, ""):
            cleaned_line = strip_ansi(line)
            console_output += cleaned_line
            # Update chat history and terminal output dynamically
            history[-1] = (prompt, f"🤖 Agent is working...\n\n```\n{console_output[-4000:]}\n```")
            yield history, console_output, gr.update()

        proc.stdout.close()
        return_code = proc.wait()

        if return_code == 0:
            status_msg = "✅ Qwen Code Agent finished successfully!"
        else:
            status_msg = f"❌ Qwen Code Agent exited with return code {return_code}."

        history[-1] = (prompt, f"{status_msg}\n\n### Terminal Log Summary:\n```\n{console_output[-4000:]}\n```")
        yield history, console_output, gr.update(choices=list_workspace_files())

    except Exception as e:
        error_msg = f"Failed to execute command: {str(e)}"
        history[-1] = (prompt, f"❌ Error: {error_msg}")
        yield history, error_msg, gr.update()

# UI Layout
with gr.Blocks(theme=gr.themes.Soft(primary_hue="purple", secondary_hue="blue")) as demo:
    gr.HTML("""
    <div style='text-align: center; margin-bottom: 20px;'>
        <h1 style='color: #8A2BE2; font-size: 2.5rem; margin-bottom: 5px;'>🤖 Qwen-Code3 AI Coding Workspace</h1>
        <p style='font-size: 1.1rem; color: #555;'>The ultimate terminal coding agent powered by Qwen. Connect your API key and let Qwen-Code code for you!</p>
    </div>
    """)

    with gr.Accordion("⚙️ API Configuration (Credentials are kept in-memory and never saved to disk)", open=True):
        with gr.Row():
            provider = gr.Dropdown(
                label="API Provider",
                choices=[
                    "DashScope (Qwen-compatible)",
                    "Alibaba Cloud Coding Plan (Beijing)",
                    "Alibaba Cloud Coding Plan (International)",
                    "OpenAI",
                    "Anthropic Claude",
                    "Google Gemini"
                ],
                value="DashScope (Qwen-compatible)"
            )
            api_key = gr.Textbox(
                label="API Key",
                placeholder="Enter your sk-... or other provider credentials",
                type="password"
            )
            model_name = gr.Dropdown(
                label="Model",
                choices=["qwen3-coder-plus", "qwen3.5-plus", "gpt-4o", "claude-3-5-sonnet-latest", "gemini-1.5-pro", "custom"],
                value="qwen3-coder-plus"
            )
            custom_model = gr.Textbox(
                label="Custom Model ID (if selected 'custom')",
                placeholder="e.g. qwen-max",
                visible=False
            )

    # Automatically show/hide custom model input
    def update_model_visibility(model_choice):
        return gr.update(visible=(model_choice == "custom"))
    model_name.change(update_model_visibility, inputs=[model_name], outputs=[custom_model])

    with gr.Tabs():
        with gr.Tab("💬 Chat & Agent Agentic Terminal"):
            with gr.Row():
                with gr.Column(scale=3):
                    chatbot = gr.Chatbot(label="Agent Workspace Log", bubble_full_width=False, height=500)
                    prompt_input = gr.Textbox(
                        label="What should the Qwen Code Agent build or fix?",
                        placeholder="e.g. 'Build a python server in main.py that returns health status'",
                        lines=2
                    )
                    run_btn = gr.Button("🚀 Run Qwen Code Agent", variant="primary")
                with gr.Column(scale=2):
                    console_log = gr.Textbox(
                        label="📟 Live Terminal Stdout (Scrolls with Agent progress)",
                        placeholder="Stdout will stream here...",
                        interactive=False,
                        lines=25,
                        max_lines=30
                    )

        with gr.Tab("📂 Workspace File Explorer"):
            with gr.Row():
                with gr.Column(scale=1):
                    file_list = gr.Dropdown(
                        label="Workspace Files",
                        choices=list_workspace_files(),
                        interactive=True
                    )
                    refresh_files_btn = gr.Button("🔄 Refresh Files")
                    download_btn = gr.Button("📥 Download Workspace (.ZIP)", variant="secondary")
                    download_file_output = gr.File(label="Workspace Download Link")
                with gr.Column(scale=3):
                    file_path_display = gr.Textbox(label="Editing File Path", interactive=False)
                    file_content_editor = gr.Code(label="File Content Editor / Viewer", language="python", lines=20)
                    save_file_btn = gr.Button("💾 Save Changes", variant="primary")
                    save_status = gr.Markdown()

            # Explorer actions
            def on_file_selected(filepath):
                if not filepath:
                    return "", ""
                return filepath, load_file_content(filepath)
            
            file_list.change(on_file_selected, inputs=[file_list], outputs=[file_path_display, file_content_editor])
            
            refresh_files_btn.click(
                lambda: gr.update(choices=list_workspace_files()),
                outputs=[file_list]
            )
            
            save_file_btn.click(
                save_file_content,
                inputs=[file_path_display, file_content_editor],
                outputs=[save_status, file_list]
            )
            
            download_btn.click(
                download_workspace,
                outputs=[download_file_output]
            )

        with gr.Tab("📋 Workspace Templates"):
            gr.Markdown("### Initialize your Workspace with a sample project template:")
            with gr.Row():
                py_template_btn = gr.Button("🐍 Python Math Library Template", variant="secondary")
                node_template_btn = gr.Button("🟢 NodeJS Web App Template", variant="secondary")
                html_template_btn = gr.Button("🎨 HTML Landing Page Template", variant="secondary")
            
            template_status = gr.Markdown()
            
            py_template_btn.click(
                lambda: load_template("Python Math Library"),
                outputs=[template_status, file_list]
            )
            node_template_btn.click(
                lambda: load_template("NodeJS Web App"),
                outputs=[template_status, file_list]
            )
            html_template_btn.click(
                lambda: load_template("HTML Landing Page"),
                outputs=[template_status, file_list]
            )

    # Connect Run Button
    run_btn.click(
        run_agent,
        inputs=[provider, api_key, model_name, custom_model, prompt_input, chatbot],
        outputs=[chatbot, console_log, file_list]
    )

if __name__ == "__main__":
    demo.queue()
    demo.launch(server_name="0.0.0.0", server_port=7860)