Update app.py

app.py

@@ -1,75 +1,79 @@
-import re
-import json
-import os
-
 import gradio as gr
-from PIL import Image, ImageFilter
+import requests
+import json
 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
-
 # — USER INFO & MODEL LISTING —
 
-def show_profile(profile):
-    return f"✅ Logged in as **{profile.username}**" if profile else "*Not logged in.*"
+def show_profile(profile: gr.OAuthProfile | None) -> str:
+    if profile is None:
+        return "*Not logged in.*"
+    return f"✅ Logged in as **{profile.username}**"
 
-def list_private_models(profile, token):
-    if not (profile and 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'})"
-        for m in list_models(author=profile.username, token=token.token)
+        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)
 
 # — BUTTON‑ENABLING HELPERS —
 
-def enable_create(profile, token):
-    return gr.update(interactive=bool(profile and token))
+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, token):
-    return gr.update(interactive=bool(repo_id and profile and 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))
 
 # — CORE ACTIONS —
 
-def create_space(repo_name, sdk, profile, token):
-    if not (profile and token):
+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>"
-    rid = f"{profile.username}/{repo_name.replace(' ', '-') }"
+    repo_id = f"{profile.username}/{repo_name}"
     create_repo(
-        repo_id=rid,
-        token=token.token,
+        repo_id=repo_id,
+        token=oauth_token.token,
         exist_ok=True,
         repo_type="space",
         space_sdk=sdk
     )
-    url    = f"https://huggingface.co/spaces/{rid}"
-    log    = f"✅ Space ready: {url} (SDK: {sdk})"
-    iframe = f'<iframe src="{url}" width="100%" height="400px"></iframe>'
-    return rid, log, iframe
-
-def write_file(path, content, repo_id, profile, token):
-    if not (profile and token):
-        return "⚠️ Please log in first."
-    if not repo_id:
-        return "⚠️ Please create a Space first."
-    with open(path, "w") as f:
-        f.write(content)
-    upload_file(
-        path_or_fileobj=path,
-        path_in_repo=path,
-        repo_id=repo_id,
-        token=token.token,
-        repo_type="space"
-    )
-    return f"✅ Wrote and uploaded `{path}`"
-
-def upload_file_to_space(file, path, repo_id, profile, token):
-    if not (profile and token):
+    url    = f"https://huggingface.co/spaces/{repo_id}"
+    logmsg = f"✅ Space ready: {url} (SDK: {sdk})"
+    iframe = f'<iframe src="{url}" width="100%" height="500px"></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 "⚠️ Please create a Space first."
@@ -77,223 +81,148 @@ def upload_file_to_space(file, path, repo_id, profile, token):
         return "⚠️ No file selected."
     upload_file(
         path_or_fileobj=file.name,
-        path_in_repo=path,
+        path_in_repo=path_in_repo,
         repo_id=repo_id,
-        token=token.token,
+        token=oauth_token.token,
         repo_type="space"
     )
-    return f"✅ Uploaded `{path}` to `{repo_id}`"
+    return f"✅ Uploaded `{path_in_repo}` to `{repo_id}`"
 
-def _fetch_space_logs_level(repo_id, level):
-    r = get_session().get(
-        f"{constants.ENDPOINT}/api/spaces/{repo_id}/jwt",
-        headers=build_hf_headers()
-    )
+def _fetch_space_logs_level(repo_id: str, level: str) -> str:
+    # 1) Get SSE JWT
+    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}"
+    # 2) Stream logs
+    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:].decode())
-                lines.append(f"[{ev.get('timestamp','')}] {ev.get('data','')}")
+            if not raw.startswith(b"data: "):
+                continue
+            payload = raw[len(b"data: "):]
+            try:
+                event = json.loads(payload.decode())
+            except json.JSONDecodeError:
+                continue
+            ts  = event.get("timestamp", "")
+            txt = event.get("data", "")
+            lines.append(f"[{ts}] {txt}")
     return "\n".join(lines)
 
-def get_build_logs(repo_id, profile, token):
-    if not (profile and token and repo_id):
+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 "⚠️ Please log in and create a Space first."
     return _fetch_space_logs_level(repo_id, "build")
 
-def get_container_logs(repo_id, profile, token):
-    if not (profile and token and repo_id):
+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")
 
-# — FUNCTION DECLARATIONS & TOOL —
-
-func_decls = [
-    types.FunctionDeclaration(
-        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"]
-        }
-    ),
-    types.FunctionDeclaration(
-        name="write_file",
-        description="Write/overwrite a file in the Space",
-        parameters={
-            "type": "object",
-            "properties": {
-                "path":    {"type": "string"},
-                "content": {"type": "string"}
-            },
-            "required": ["path", "content"]
-        }
-    ),
-    types.FunctionDeclaration(
-        name="list_files",
-        description="List files in the Space",
-        parameters={
-            "type": "object",
-            "properties": {"repo_id": {"type": "string"}},
-            "required": ["repo_id"]
-        }
-    ),
-    types.FunctionDeclaration(
-        name="get_build_logs",
-        description="Fetch build logs",
-        parameters={
-            "type": "object",
-            "properties": {"repo_id": {"type": "string"}},
-            "required": ["repo_id"]
-        }
-    ),
-    types.FunctionDeclaration(
-        name="get_run_logs",
-        description="Fetch container logs",
-        parameters={
-            "type": "object",
-            "properties": {"repo_id": {"type": "string"}},
-            "required": ["repo_id"]
-        }
-    ),
-]
-tool = types.Tool(function_declarations=func_decls)
-
-# — CHAT HANDLER —
-
-def execute_function_by_name(name, args, profile, token, session):
-    if name == "create_space":
-        rid, log, iframe = create_space(
-            args["repo_name"], session["sdk"], profile, token
-        )
-        session["repo_id"] = rid
-        session["iframe"] = iframe
-        session["log"] = log
-        return {"log": log, "iframe": iframe}
-    if name == "write_file":
-        status = write_file(
-            args["path"], args["content"],
-            session["repo_id"], profile, token
-        )
-        return {"status": status}
-    if name == "list_files":
-        fl = list_repo_files(
-            session["repo_id"], token=token.token, repo_type="space"
-        )
-        return {"files": "\n".join(fl)}
-    if name == "get_build_logs":
-        return {"log": get_build_logs(
-            session["repo_id"], profile, token
-        )}
-    if name == "get_run_logs":
-        return {"log": get_container_logs(
-            session["repo_id"], profile, token
-        )}
-    return {"error": f"Unknown function {name}"}
+# — BUILD THE UI —
 
+with gr.Blocks(title="HF OAuth + Space Manager with Logs") as demo:
+    gr.Markdown(
+        "## Sign in with Hugging Face + Manage Your Space\n\n"
+        "1. Sign in\n"
+        "2. Create a Space (Gradio/Streamlit)\n"
+        "3. Upload files to it\n"
+        "4. Fetch build and container logs\n\n"
+        "---"
+    )
 
-def process_message(
-    profile, token, user_msg,
-    gemini_key, sidebar_repo, sidebar_sdk,
-    chat_history, session
-):
-    if session.get("chat") is None:
-        client = genai.Client(api_key=gemini_key)
-        model = client.chats.create(
-            model="gemini-2.5-flash-preview-04-17",
-            config=types.GenerateContentConfig(
-                system_instruction=(
-                    "You are a HF Spaces admin. You can create spaces, "
-                    "write files, list files, and fetch logs."
-                ),
-                temperature=0,
-                tools=[tool],
-            ),
-        )
-        session.update({
-            "chat": model,
-            "repo_id": None,
-            "sdk": sidebar_sdk,
-            "messages": []
-        })
-    session["sdk"] = sidebar_sdk
-    if session["repo_id"] is None:
-        m = re.search(r"(?:call it|called)\s+([\w \-]+)", user_msg, re.IGNORECASE)
-        if m:
-            args = {"repo_name": m.group(1), "sdk": sidebar_sdk}
-            result = execute_function_by_name("create_space", args, profile, token, session)
-            session["messages"].append({"role": "assistant", "content": result["log"]})
-            return session["messages"], session["iframe"], session["log"], "", session
-    session["messages"].append({"role": "user", "content": user_msg})
-    response = session["chat"].send_message(user_msg)
-    while True:
-        part = response.candidates[0].content.parts[0]
-        if not part.function_call:
-            break
-        fname = part.function_call.name
-        fargs = part.function_call.args or {}
-        result = execute_function_by_name(fname, fargs, profile, token, session)
-        resp_part = types.Part.from_function_response(name=fname, response=result)
-        response = session["chat"].send_message(resp_part)
-    assistant_text = response.text
-    session["messages"].append({"role": "assistant", "content": assistant_text})
-    return (
-        session["messages"],
-        session.get("iframe", ""),
-        session.get("log", ""),
-        session.get("files", ""),
-        session
+    # — LOGIN & MODEL LIST —
+    login_btn = gr.LoginButton(variant="huggingface", size="lg")
+    status_md = gr.Markdown("*Not logged in.*")
+    models_md = gr.Markdown()
+    demo.load(show_profile,         inputs=None, outputs=status_md)
+    login_btn.click(show_profile,   inputs=None, outputs=status_md)
+    demo.load(list_private_models,  inputs=None, outputs=models_md)
+    login_btn.click(list_private_models,
+                    inputs=None,     outputs=models_md)
+
+    # — CREATE SPACE —
+    repo_name    = gr.Textbox(label="New Space name", placeholder="my-space")
+    sdk_selector = gr.Radio(
+        choices=["gradio","streamlit"],
+        value="gradio",
+        label="Space template (SDK)"
+    )
+    create_btn   = gr.Button("Create Space", interactive=False)
+    session_id   = gr.Textbox(visible=False)
+    create_logs  = gr.Textbox(label="Create Logs", interactive=False, lines=3)
+    preview      = gr.HTML("<p>No Space created yet.</p>")
+
+    demo.load(enable_create,        inputs=None,        outputs=[create_btn])
+    login_btn.click(enable_create,  inputs=None,        outputs=[create_btn])
+
+    create_btn.click(
+        fn=create_space,
+        inputs=[repo_name, sdk_selector],
+        outputs=[session_id, create_logs, preview]
     )
 
+    # — UPLOAD FILES —
+    path_in_repo  = gr.Textbox(label="Path in Space", value="app.py")
+    file_uploader = gr.File(label="Select file")
+    upload_btn    = gr.Button("Upload File", interactive=False)
+    upload_logs   = gr.Textbox(label="Upload Logs", interactive=False, lines=2)
+
+    demo.load(enable_repo_actions,
+              inputs=[session_id],
+              outputs=[upload_btn])
+    login_btn.click(enable_repo_actions,
+                    inputs=[session_id],
+                    outputs=[upload_btn])
+    session_id.change(enable_repo_actions,
+                      inputs=[session_id],
+                      outputs=[upload_btn])
+
+    upload_btn.click(
+        fn=upload_file_to_space,
+        inputs=[file_uploader, path_in_repo, session_id],
+        outputs=[upload_logs]
+    )
 
-def sync_manual(profile, token, session):
-    if not (profile and token and session.get("repo_id")):
-        return "", "⚠️ Cannot sync manual changes.", "", session
-    files = list_repo_files(session["repo_id"], token=token.token, repo_type="space")
-    log   = "🔄 Manual changes synced."
-    return session.get("iframe", ""), log, "\n".join(files), session
+    # — FETCH BUILD & CONTAINER LOGS —
+    build_logs_btn     = gr.Button("Get Build Logs", interactive=False)
+    container_logs_btn = gr.Button("Get Container Logs", interactive=False)
+    build_logs_md      = gr.Textbox(label="Build Logs", interactive=False, lines=10)
+    container_logs_md  = gr.Textbox(label="Container Logs", interactive=False, lines=10)
+
+    # enable both log buttons
+    for btn in (build_logs_btn, container_logs_btn):
+        demo.load(enable_repo_actions,
+                  inputs=[session_id],
+                  outputs=[btn])
+        login_btn.click(enable_repo_actions,
+                        inputs=[session_id],
+                        outputs=[btn])
+        session_id.change(enable_repo_actions,
+                          inputs=[session_id],
+                          outputs=[btn])
+
+    build_logs_btn.click(
+        fn=get_build_logs,
+        inputs=[session_id],
+        outputs=[build_logs_md]
+    )
+    container_logs_btn.click(
+        fn=get_container_logs,
+        inputs=[session_id],
+        outputs=[container_logs_md]
+    )
 
-with gr.Blocks(css="""
-#sidebar { background:#f2f2f2; padding:1rem; border-right:1px solid #ccc; }
-#main    { padding:1rem; }
-""") as demo:
-    with gr.Column(elem_id="sidebar", scale=1):
-        gr.Markdown("### 🔑 HF Login & Config")
-        login_btn     = gr.LoginButton(variant="huggingface", size="sm")
-        profile_state = gr.State(None)
-        token_state   = gr.State(None)
-        login_btn.click(None, [], [profile_state, token_state])
-        status_md = gr.Markdown("*Not logged in.*")
-        profile_state.change(show_profile, [profile_state], [status_md])
-        models_md = gr.Markdown()
-        profile_state.change(list_private_models,[profile_state, token_state],[models_md])
-        gemini_key   = gr.Textbox(label="Gemini API Key", type="password")
-        sidebar_repo = gr.Textbox(label="Space name", placeholder="blurrtest")
-        sidebar_sdk  = gr.Radio(["gradio","streamlit"],value="gradio",label="SDK")
-        gr.Markdown("---")
-        confirm_btn  = gr.Button("🔄 Sync Manual Changes")
-    with gr.Column(elem_id="main", scale=3):
-        tabs = gr.Tabs()
-        with tabs:
-            with gr.TabItem("💬 Chat"):
-                chatbox = gr.Chatbot(type="messages")
-                user_input = gr.Textbox(show_label=False, placeholder="Generate code…")
-                send_btn = gr.Button("Send")
-            with gr.TabItem("🛠️ Manual"):
-                gr.Markdown("#### Create / Rename Space")
-                repo_m = gr.Textbox(label="Name", value="")
-                sdk_m = gr.Radio(["gradio","streamlit"],value="gradio",label="SDK")
-                create_btn = gr.Button("Create Space")
-                sess_id = gr.Textbox(visible=False)
-                log_c = gr.Textbox(label="Log",interactive=False,lines=2)
-                preview = gr.HTML("<p>No Space yet.</p>")
-                create_btn.click(create_space,[repo_m,sdk_m,profile_state,token
+if __name__ == "__main__":
+    demo.launch()