Update app.py

app.py

@@ -1,19 +1,21 @@
-import json, os
+import os, json, requests
 import gradio as gr
 from huggingface_hub import (
-    create_repo, list_models, upload_file, list_repo_files, constants
+    create_repo, list_models, upload_file, 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 — #
+# — USER INFO & MODEL LISTING — (unchanged)
+def show_profile(profile: gr.OAuthProfile | None) -> str:
+    if profile is None:
+        return "*Not logged in.*"
+    return f"✅ Logged in as **{profile.username}**"
 
-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):
+def list_private_models(profile: gr.OAuthProfile | None,
+                        oauth_token: gr.OAuthToken | None) -> str:
+    if profile is None or oauth_token is None:
         return "Please log in to see your models."
     models = [
         f"{m.id} ({'private' if m.private else 'public'})"
@@ -21,60 +23,94 @@ def list_private_models(profile, oauth_token):
     ]
     return "No models found." if not models else "Models:\n\n" + "\n - ".join(models)
 
-# — MANUAL HELPERS — #
-
-def enable_create(profile, oauth_token):
-    return gr.update(interactive=bool(profile and oauth_token))
+# — BUTTON‑ENABLING HELPERS — (unchanged)
+def enable_create(profile: gr.OAuthProfile | None,
+                  oauth_token: gr.OAuthToken | None):
+    return gr.update(interactive=profile is not None)
 
-def enable_repo_actions(repo_id, profile, oauth_token):
+def enable_repo_actions(repo_id: str,
+                        profile: gr.OAuthProfile | None,
+                        oauth_token: gr.OAuthToken | None):
     return gr.update(interactive=bool(repo_id and profile and oauth_token))
 
-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):
+# — CORE ACTIONS — (unchanged)
+def create_space(repo_name: str, sdk: str,
+                 profile: gr.OAuthProfile | None,
+                 oauth_token: gr.OAuthToken | None
+) -> tuple[str, str, str]:
+    if not profile or not oauth_token:
+        return "", "⚠️ Please log in first.", "<p>No Space created yet.</p>"
+    repo_id = f"{profile.username}/{repo_name}"
+    create_repo(
+        repo_id=repo_id,
+        token=oauth_token.token,
+        exist_ok=True,
+        repo_type="space",
+        space_sdk=sdk
+    )
+    url    = f"https://huggingface.co/spaces/{repo_id}"
+    logmsg = f"✅ Space ready: {url} (SDK: {sdk})"
+    iframe = f'<iframe src="{url}" width="100%" height="400px"></iframe>'
+    return repo_id, logmsg, iframe
+
+def upload_file_to_space(file, path_in_repo: str, repo_id: str,
+                         profile: gr.OAuthProfile | None,
+                         oauth_token: gr.OAuthToken | None
+) -> str:
+    if not profile or not oauth_token:
         return "⚠️ Please log in first."
     if not repo_id:
-        return "⚠️ Create a Space first."
+        return "⚠️ Please 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())
+    upload_file(
+        path_or_fileobj=file.name,
+        path_in_repo=path_in_repo,
+        repo_id=repo_id,
+        token=oauth_token.token,
+        repo_type="space"
+    )
+    return f"✅ Uploaded `{path_in_repo}` to `{repo_id}`"
+
+def _fetch_space_logs_level(repo_id: str, level: str) -> str:
+    jwt_url = f"{constants.ENDPOINT}/api/spaces/{repo_id}/jwt"
+    r = get_session().get(jwt_url, headers=build_hf_headers())
     hf_raise_for_status(r)
     jwt = r.json()["token"]
-    url = f"https://api.hf.space/v1/{repo_id}/logs/{level}"
+    logs_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:
+    with get_session().get(logs_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','')}")
+                ev = json.loads(raw[len(b"data: "):].decode())
+                ts, txt = ev.get("timestamp", ""), ev.get("data", "")
+                lines.append(f"[{ts}] {txt}")
     return "\n".join(lines)
 
-def list_files(repo_id, profile, oauth_token):
+def get_build_logs(repo_id: str,
+                   profile: gr.OAuthProfile | None,
+                   oauth_token: gr.OAuthToken | None
+) -> str:
     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"))
+        return "⚠️ Please log in and create a Space first."
+    return _fetch_space_logs_level(repo_id, "build")
+
+def get_container_logs(repo_id: str,
+                       profile: gr.OAuthProfile | None,
+                       oauth_token: gr.OAuthToken | None
+) -> str:
+    if not (profile and oauth_token and repo_id):
+        return "⚠️ Please log in and create a Space first."
+    return _fetch_space_logs_level(repo_id, "run")
 
-# — GEMINI FUNCTION DECLARATIONS — #
 
+# — GEMINI FUNCTION DECLARATIONS — 
 func_decls = [
     {
-      "name":"create_space","description":"Create/get a HF Space",
-      "parameters":{
+      "name": "create_space",
+      "description": "Create or get a Hugging Face Space",
+      "parameters": {
         "type":"object",
         "properties":{
           "repo_name":{"type":"string"},
@@ -84,7 +120,8 @@ func_decls = [
       }
     },
     {
-      "name":"list_files","description":"List files in Space",
+      "name":"list_files",
+      "description":"List files in the Space",
       "parameters":{
         "type":"object",
         "properties":{"repo_id":{"type":"string"}},
@@ -92,7 +129,8 @@ func_decls = [
       }
     },
     {
-      "name":"get_build_logs","description":"Fetch build logs",
+      "name":"get_build_logs",
+      "description":"Fetch the build logs",
       "parameters":{
         "type":"object",
         "properties":{"repo_id":{"type":"string"}},
@@ -100,7 +138,8 @@ func_decls = [
       }
     },
     {
-      "name":"get_run_logs","description":"Fetch run logs",
+      "name":"get_run_logs",
+      "description":"Fetch the run logs",
       "parameters":{
         "type":"object",
         "properties":{"repo_id":{"type":"string"}},
@@ -109,18 +148,22 @@ func_decls = [
     },
 ]
 
-# — CHAT HANDLER — #
-
-def process_message(profile, oauth_token, user_msg, gemini_key, repo_name, sdk, chat_history, session):
-    # initialize on first call
+# — CHAT HANDLER — 
+def process_message(profile, oauth_token, user_msg,
+                    gemini_key, repo_name, sdk,
+                    chat_history, session):
+    # initialize chat 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.",
+            system_instruction=(
+                "You are a HF Spaces admin. "
+                "Use functions to create spaces, list files, or fetch logs."
+            ),
             temperature=0,
             tools=[ types.Tool(function_declarations=func_decls) ]
         )
-        session["chat"] = client.chats.create(model="gemini-2.0-flash", config=cfg)
+        session["chat"]    = client.chats.create(model="gemini-2.0-flash", config=cfg)
         session["repo_id"] = None
 
     chat = session["chat"]
@@ -133,27 +176,35 @@ def process_message(profile, oauth_token, user_msg, gemini_key, repo_name, sdk,
         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)
+            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}
+            files = list_repo_files(
+                session["repo_id"], token=oauth_token.token, repo_type="space"
+            )
+            result = {"files": "\n".join(files)}
         elif name == "get_build_logs":
-            log = fetch_logs(session["repo_id"], profile, oauth_token, "build")
-            result = {"log": log}
+            result = {
+                "log": get_build_logs(session["repo_id"], profile, oauth_token)
+            }
         elif name == "get_run_logs":
-            log = fetch_logs(session["repo_id"], profile, oauth_token, "run")
-            result = {"log": log}
+            result = {
+                "log": get_container_logs(session["repo_id"], profile, oauth_token)
+            }
         else:
             result = {"log": f"⚠️ Unknown function {name}"}
 
-        # feed back
+        # feed function result back into the chat
         chat.send_message(
             types.Content(
                 role="function",
-                parts=[ types.Part(function_call=part.function_call,
-                                   function_response=json.dumps(result)) ]
+                parts=[ types.Part(
+                    function_call=part.function_call,
+                    function_response=json.dumps(result)
+                )]
             )
         )
         final = chat.get_history()[-1].parts[0].text
@@ -166,108 +217,149 @@ def process_message(profile, oauth_token, user_msg, gemini_key, repo_name, sdk,
     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
-
-# — SYNC MANUAL CHANGES — #
+    return (chat_history,
+            session.get("iframe",""),
+            session.get("log",""),
+            session.get("files",""),
+            session)
 
+# — SYNC MANUAL CHANGES — 
 def sync_manual(profile, oauth_token, session):
     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
+        return (session.get("iframe",""),
+                "⚠️ Cannot sync manual changes.",
+                session.get("files",""),
+                session)
+    files = list_repo_files(
+        session["repo_id"], token=oauth_token.token, repo_type="space"
+    )
+    session["files"] = "\n".join(files)
     session["log"]   = "🔄 Manual changes synced."
-    return session.get("iframe",""), session["log"], session["files"], session
+    return (session.get("iframe",""),
+            session["log"],
+            session["files"],
+            session)
 
-# — BUILD THE UI — #
 
+# — BUILD THE UI — 
 with gr.Blocks(css="""
-#sidebar {background:#f7f7f7;padding:1rem;border-right:1px solid #ddd;}
-#main    {padding:1rem;}
+#sidebar { background:#fafafa; padding:1em; border-right:1px solid #ddd; }
+#main    { padding:1em; }
 """) as demo:
+
     with gr.Row():
-        # Sidebar
+        # ◀ Sidebar
         with gr.Column(elem_id="sidebar", scale=1):
-            gr.Markdown("### 🔑 Login & Config")
-            hf_login = gr.LoginButton(variant="huggingface", size="sm")
-            status   = gr.Markdown("*Not logged in.*")
-            models   = gr.Markdown()
-
-            hf_login.click(show_profile, inputs=[hf_login], outputs=[status])
-            hf_login.click(list_private_models, inputs=[hf_login], outputs=[models])
-
-            gemini_key = gr.Textbox(label="Gemini API Key", type="password")
-            repo_name  = gr.Textbox(label="Space name", placeholder="my-space")
-            sdk_choice = gr.Radio(["gradio","streamlit"], value="gradio", label="SDK")
+            gr.Markdown("### 🔑 HF Login & Config")
+            hf_login     = gr.LoginButton(variant="huggingface", size="sm")
+            status_md    = gr.Markdown("*Not logged in.*")
+            models_md    = gr.Markdown()
+
+            # login callbacks
+            hf_login.click(show_profile,
+                           inputs=[hf_login], outputs=[status_md])
+            hf_login.click(list_private_models,
+                           inputs=[hf_login, hf_login], outputs=[models_md])
+
+            gemini_key   = gr.Textbox(label="Gemini API Key",
+                                      type="password")
+            sidebar_repo = gr.Textbox(label="Space name",
+                                      placeholder="e.g. my-space")
+            sidebar_sdk  = gr.Radio(["gradio","streamlit"],
+                                    value="gradio", label="SDK")
 
             gr.Markdown("---")
-            sync_btn = gr.Button("🔄 Confirm Manual Changes")
+            confirm_btn  = gr.Button("🔄 Confirm Manual Changes")
 
-        # Main area
+        # ▶ Main area: Chat & Manual Controls
         with gr.Column(elem_id="main", scale=3):
             tabs = gr.Tabs()
-
             with tabs:
                 with gr.TabItem("💬 Chat"):
-                    chatbox = gr.Chatbot(type="messages", label="Chat")
-                    user_in = gr.Textbox(show_label=False, placeholder="Ask the LLM…")
-                    send    = gr.Button("Send")
+                    chatbox   = gr.Chatbot(type="messages")
+                    user_input= gr.Textbox(show_label=False,
+                                           placeholder="Ask the LLM…")
+                    send_btn  = gr.Button("Send")
                 with gr.TabItem("🛠️ Manual"):
                     gr.Markdown("#### Create a Space")
                     repo_m = gr.Textbox(label="Name")
-                    sdk_m  = gr.Radio(["gradio","streamlit"], value="gradio", label="SDK")
-                    create_btn = gr.Button("Create Space", 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>")
-
-                    hf_login.change(enable_create, inputs=[hf_login], outputs=[create_btn])
+                    sdk_m  = gr.Radio(["gradio","streamlit"],
+                                      value="gradio", label="SDK")
+                    create_btn = gr.Button("Create Space",
+                                           interactive=False)
+                    sess_id    = gr.Textbox(visible=False)
+                    log_c      = gr.Textbox(label="Log",
+                                           interactive=False, lines=2)
+                    iframe_m   = gr.HTML("<p>No Space yet.</p>")
+
+                    hf_login.click(enable_create,
+                                   inputs=[hf_login, hf_login],
+                                   outputs=[create_btn])
                     create_btn.click(create_space,
-                                     inputs=[repo_m, sdk_m, hf_login, hf_login],
-                                     outputs=[sess, log_c, preview])
+                                     inputs=[repo_m, sdk_m,
+                                             hf_login, hf_login],
+                                     outputs=[sess_id, log_c,
+                                              iframe_m])
 
                     gr.Markdown("#### Upload File")
-                    path   = gr.Textbox(label="Path in Repo", value="app.py")
-                    file_u = gr.File()
-                    up_btn = gr.Button("Upload File", interactive=False)
-                    log_u  = gr.Textbox(label="Log", interactive=False, lines=2)
-
-                    hf_login.change(enable_repo_actions,
-                                    inputs=[sess, hf_login, hf_login],
-                                    outputs=[up_btn])
+                    path    = gr.Textbox(label="Path in Repo",
+                                         value="app.py")
+                    file_u  = gr.File()
+                    up_btn  = gr.Button("Upload File",
+                                        interactive=False)
+                    log_u   = gr.Textbox(label="Log",
+                                         interactive=False, lines=2)
+
+                    hf_login.click(enable_repo_actions,
+                                   inputs=[sess_id, hf_login, hf_login],
+                                   outputs=[up_btn])
                     up_btn.click(upload_file_to_space,
-                                 inputs=[file_u, path, sess, hf_login, hf_login],
+                                 inputs=[file_u, path, sess_id,
+                                         hf_login, hf_login],
                                  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)
-
-                    hf_login.change(enable_repo_actions,
-                                    inputs=[sess, hf_login, hf_login],
-                                    outputs=[b_btn, r_btn])
-                    b_btn.click(fetch_logs,
-                                inputs=[sess, hf_login, hf_login, gr.State("build")],
-                                outputs=[log_b])
-                    r_btn.click(fetch_logs,
-                                inputs=[sess, hf_login, hf_login, gr.State("run")],
-                                outputs=[log_r])
-
-            # Persistent panels
+                    build_btn = gr.Button("Build Logs",
+                                          interactive=False)
+                    run_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)
+
+                    hf_login.click(enable_repo_actions,
+                                   inputs=[sess_id, hf_login, hf_login],
+                                   outputs=[build_btn, run_btn])
+                    build_btn.click(get_build_logs,
+                                    inputs=[sess_id,
+                                            hf_login, hf_login],
+                                    outputs=[log_b])
+                    run_btn.click(get_container_logs,
+                                  inputs=[sess_id,
+                                          hf_login, hf_login],
+                                  outputs=[log_r])
+
+            # persistent bottom panels
             gr.Markdown("---")
             iframe_out = gr.HTML(label="🖼️ Preview")
-            log_out    = gr.Textbox(label="📋 Latest Log", lines=4)
+            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=[hf_login, hf_login, user_in, gemini_key, repo_name, sdk_choice, chatbox, state],
-               outputs=[chatbox, iframe_out, log_out, files_out, state])
-
-    sync_btn.click(sync_manual,
-                   inputs=[hf_login, hf_login, state],
-                   outputs=[iframe_out, log_out, files_out, state])
+    send_btn.click(process_message,
+                   inputs=[hf_login, hf_login,
+                           user_input, gemini_key,
+                           sidebar_repo, sidebar_sdk,
+                           chatbox, state],
+                   outputs=[chatbox, iframe_out,
+                            log_out, files_out, state])
+
+    confirm_btn.click(sync_manual,
+                      inputs=[hf_login, hf_login, state],
+                      outputs=[iframe_out, log_out,
+                               files_out, state])
 
 if __name__ == "__main__":
     demo.launch()