Create hfmaker.py

hfmaker.py

@@ -0,0 +1,228 @@
+import gradio as gr
+import requests
+import json
+from huggingface_hub import (
+    create_repo,
+    list_models,
+    upload_file,
+    constants,
+)
+from huggingface_hub.utils import build_hf_headers, get_session, hf_raise_for_status
+
+# — USER INFO & MODEL LISTING —
+
+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: 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=oauth_token.token)
+    ]
+    return "No models found." if not models else "Models:\n\n" + "\n - ".join(models)
+
+# — BUTTON‑ENABLING HELPERS —
+
+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: 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: 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="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."
+    if not file:
+        return "⚠️ No file selected."
+    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:
+    # 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"]
+    # 2) Stream logs
+    logs_url = f"https://api.hf.space/v1/{repo_id}/logs/{level}"
+    lines = []
+    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 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: 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: 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")
+
+# — 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"
+        "---"
+    )
+
+    # — 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]
+    )
+
+    # — 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]
+    )
+
+if __name__ == "__main__":
+    demo.launch()