app.py

import os, json
import gradio as gr
from huggingface_hub import (
    create_repo, list_models, upload_file, list_repo_files, constants
)
from huggingface_hub.utils import build_hf_headers, get_session, hf_raise_for_status
from google import genai
from google.genai import types

# — HF LOGIN CALLBACKS — #

def show_profile(profile):
    return f"✅ Logged in as **{profile.username}**" if profile else "*Not logged in.*"

def list_private_models(profile, oauth_token):
    if not (profile and oauth_token):
        return "Please log in to see your models."
    models = [
        f"{m.id} ({'private' if m.private else 'public'})"
        for m in list_models(author=profile.username, token=oauth_token.token)
    ]
    return "No models found." if not models else "Models:\n\n" + "\n - ".join(models)

# — HF SPACE HELPERS — #

def create_space(repo_name, sdk, profile, oauth_token):
    if not (profile and oauth_token):
        return "", "⚠️ Please log in first.", "<p>No Space.</p>"
    rid = f"{profile.username}/{repo_name}"
    create_repo(rid, token=oauth_token.token, exist_ok=True, repo_type="space", space_sdk=sdk)
    url = f"https://huggingface.co/spaces/{rid}"
    return rid, f"✅ Space ready: {url}", f'<iframe src="{url}" width="100%" height="400px"></iframe>'

def upload_file_to_space(file, path, repo_id, profile, oauth_token):
    if not (profile and oauth_token):
        return "⚠️ Please log in first."
    if not repo_id:
        return "⚠️ Create a Space first."
    if not file:
        return "⚠️ No file selected."
    upload_file(file.name, path, repo_id, token=oauth_token.token, repo_type="space")
    return f"✅ Uploaded `{path}`"

def fetch_logs(repo_id, profile, oauth_token, level):
    if not (profile and oauth_token and repo_id):
        return "⚠️ Log in & create a Space first."
    # get JWT
    r = get_session().get(f"{constants.ENDPOINT}/api/spaces/{repo_id}/jwt",
                          headers=build_hf_headers())
    hf_raise_for_status(r); jwt = r.json()["token"]
    url = f"https://api.hf.space/v1/{repo_id}/logs/{level}"
    lines = []
    with get_session().get(url, headers=build_hf_headers(token=jwt), stream=True) as resp:
        hf_raise_for_status(resp)
        for raw in resp.iter_lines():
            if raw.startswith(b"data: "):
                ev = json.loads(raw[6:])
                lines.append(f"[{ev.get('timestamp','')}] {ev.get('data','')}")
    return "\n".join(lines)

def list_files(repo_id, profile, oauth_token):
    if not (profile and oauth_token and repo_id):
        return "⚠️ Log in & create a Space first."
    return "\n".join(list_repo_files(repo_id, token=oauth_token.token, repo_type="space"))

# — GEMINI FUNCTION DECLS — #

func_decls = [
    {
      "name":"create_space","description":"Create/get a HF Space",
      "parameters":{
        "type":"object",
        "properties":{
          "repo_name":{"type":"string"},
          "sdk":{"type":"string","enum":["gradio","streamlit"]}
        },
        "required":["repo_name","sdk"]
      }
    },
    {
      "name":"list_files","description":"List files in Space",
      "parameters":{
        "type":"object",
        "properties":{"repo_id":{"type":"string"}},
        "required":["repo_id"]
      }
    },
    {
      "name":"get_build_logs","description":"Fetch build logs",
      "parameters":{
        "type":"object",
        "properties":{"repo_id":{"type":"string"}},
        "required":["repo_id"]
      }
    },
    {
      "name":"get_run_logs","description":"Fetch run logs",
      "parameters":{
        "type":"object",
        "properties":{"repo_id":{"type":"string"}},
        "required":["repo_id"]
      }
    },
]

# — CHAT HANDLER — #

def process_message(profile, oauth_token, user_msg, gemini_key, repo_name, sdk, chat_history, session):
    # init on first call
    if session.get("chat") is None:
        client = genai.Client(api_key=gemini_key)
        cfg = types.GenerateContentConfig(
            system_instruction="You are an admin for HF Spaces: create spaces, list files, and fetch logs.",
            temperature=0,
            tools=[ types.Tool(function_declarations=func_decls) ]
        )
        session["chat"] = client.chats.create(model="gemini-2.0-flash", config=cfg)
        session["repo_id"] = None

    chat = session["chat"]
    chat_history.append((user_msg, None))
    resp = chat.send_message(user_msg)
    part = resp.candidates[0].content.parts[0]

    result = {}
    if part.function_call:
        name = part.function_call.name
        args = json.loads(part.function_call.args)
        if name == "create_space":
            rid, log, iframe = create_space(args["repo_name"], args["sdk"], profile, oauth_token)
            session["repo_id"] = rid
            result = {"log": log, "iframe": iframe}
        elif name == "list_files":
            files = list_files(session["repo_id"], profile, oauth_token)
            result = {"files": files}
        elif name == "get_build_logs":
            log = fetch_logs(session["repo_id"], profile, oauth_token, "build")
            result = {"log": log}
        elif name == "get_run_logs":
            log = fetch_logs(session["repo_id"], profile, oauth_token, "run")
            result = {"log": log}
        else:
            result = {"log": f"⚠️ Unknown function {name}"}

        # feed back into chat
        chat.send_message(
            types.Content(
                role="function",
                parts=[ types.Part(function_call=part.function_call,
                                   function_response=json.dumps(result)) ]
            )
        )
        final = chat.get_history()[-1].parts[0].text
    else:
        final = part.text

    # update panels
    if "iframe" in result: session["iframe"] = result["iframe"]
    if "log"    in result: session["log"]    = result["log"]
    if "files"  in result: session["files"]  = result["files"]

    chat_history[-1] = (user_msg, final)
    return chat_history, session.get("iframe",""), session.get("log",""), session.get("files",""), session

def sync_manual(profile, oauth_token, session):
    # pull in any manual file changes
    if not (profile and oauth_token and session.get("repo_id")):
        return session.get("iframe",""), "⚠️ Cannot sync.", session.get("files",""), session
    files = list_files(session["repo_id"], profile, oauth_token)
    session["files"] = files
    session["log"]   = "🔄 Manual changes synced."
    return session.get("iframe",""), session["log"], session["files"], session

# — LAYOUT — #

with gr.Blocks(css="""
#sidebar {background:#f7f7f7;padding:1em;border-right:1px solid #ddd;}
#main    {padding:1em;}
""") as demo:

    with gr.Row():
        # Sidebar
        with gr.Column(elem_id="sidebar", scale=1):
            gr.Markdown("### 🔑 Login & Config")
            login_btn     = gr.LoginButton(variant="huggingface", size="sm")
            profile_state = gr.State(None)
            token_state   = gr.State(None)
            # capture login outputs
            login_btn.click(None, outputs=[profile_state, token_state])

            status_md = gr.Markdown("*Not logged in.*")
            profile_state.change(show_profile, inputs=[profile_state], outputs=[status_md])
            login_btn.click(show_profile, inputs=[profile_state], outputs=[status_md])

            models_md = gr.Markdown()
            profile_state.change(list_private_models,
                                 inputs=[profile_state, token_state],
                                 outputs=[models_md])
            login_btn.click(list_private_models,
                            inputs=[profile_state, token_state],
                            outputs=[models_md])

            gemini_key = gr.Textbox(label="Gemini API Key", type="password")
            repo_name  = gr.Textbox(label="Space name", placeholder="e.g. my-space")
            sdk_choice = gr.Radio(["gradio","streamlit"], value="gradio", label="SDK")

            gr.Markdown("---")
            sync_btn = gr.Button("🔄 Confirm Manual Changes")
            sync_out = gr.Markdown()
            sync_btn.click(sync_manual,
                           inputs=[profile_state, token_state, gr.State({})],
                           outputs=[gr.HTML(), sync_out, gr.Textbox(), gr.State({})])

        # Main area
        with gr.Column(elem_id="main", scale=3):
            tabs = gr.Tabs()
            with tabs:
                with gr.TabItem("💬 Chat"):
                    chatbox = gr.Chatbot(type="messages")
                    user_in = gr.Textbox(show_label=False, placeholder="Ask the LLM…")
                    send    = gr.Button("Send")
                with gr.TabItem("🛠️ Manual"):
                    gr.Markdown("#### Create a Space")
                    repo_m  = gr.Textbox(label="Name")
                    sdk_m   = gr.Radio(["gradio","streamlit"], label="SDK")
                    create  = gr.Button("Create", interactive=False)
                    sess    = gr.Textbox(visible=False)
                    log_c   = gr.Textbox(label="Log", interactive=False, lines=2)
                    preview = gr.HTML("<p>No Space yet</p>")

                    profile_state.change(enable_create,
                                         inputs=[profile_state, token_state],
                                         outputs=[create])
                    login_btn.click(enable_create,
                                    inputs=[profile_state, token_state],
                                    outputs=[create])

                    create.click(create_space,
                                 inputs=[repo_m, sdk_m, profile_state, token_state],
                                 outputs=[sess, log_c, preview])

                    gr.Markdown("#### Upload File")
                    path    = gr.Textbox(label="Path in Repo", value="app.py")
                    file_u  = gr.File()
                    up_btn  = gr.Button("Upload", interactive=False)
                    log_u   = gr.Textbox(label="Log", interactive=False, lines=2)

                    profile_state.change(enable_repo_actions,
                                         inputs=[sess, profile_state, token_state],
                                         outputs=[up_btn])
                    login_btn.click(enable_repo_actions,
                                    inputs=[sess, profile_state, token_state],
                                    outputs=[up_btn])

                    up_btn.click(upload_file_to_space,
                                 inputs=[file_u, path, sess, profile_state, token_state],
                                 outputs=[log_u])

                    gr.Markdown("#### Fetch Logs")
                    b_btn = gr.Button("Build Logs", interactive=False)
                    r_btn = gr.Button("Run Logs",   interactive=False)
                    log_b = gr.Textbox(label="Build", interactive=False, lines=5)
                    log_r = gr.Textbox(label="Run",   interactive=False, lines=5)

                    profile_state.change(enable_repo_actions,
                                         inputs=[sess, profile_state, token_state],
                                         outputs=[b_btn, r_btn])
                    login_btn.click(enable_repo_actions,
                                    inputs=[sess, profile_state, token_state],
                                    outputs=[b_btn, r_btn])

                    b_btn.click(fetch_logs,
                                inputs=[sess, profile_state, token_state, gr.State("build")],
                                outputs=[log_b])
                    r_btn.click(fetch_logs,
                                inputs=[sess, profile_state, token_state, gr.State("run")],
                                outputs=[log_r])

            # always‑visible panels
            gr.Markdown("---")
            iframe_out = gr.HTML(label="🖼️ Preview")
            log_out    = gr.Textbox(label="📋 Latest Log", lines=4)
            files_out  = gr.Textbox(label="📚 Files", lines=4)

    state = gr.State({})
    send.click(process_message,
               inputs=[profile_state, token_state, user_in, gemini_key, repo_name, sdk_choice, chatbox, state],
               outputs=[chatbox, iframe_out, log_out, files_out, state])

if __name__ == "__main__":
    demo.launch()