Spaces:
Sleeping
Sleeping
| 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() | |