testtsettset / backup.py
wuhp's picture
Create backup.py
b874ac3 verified
raw
history blame contribute delete
7.64 kB
import re
import json
import time
import importlib.metadata
import gradio as gr
from huggingface_hub import create_repo, upload_file, list_models, constants
from huggingface_hub.utils import build_hf_headers, get_session, hf_raise_for_status
from google import genai
from google.genai.types import Tool, GenerateContentConfig, GoogleSearch
# — 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)
# — UTILITIES —
def get_sdk_version(sdk_choice: str) -> str:
pkg = "gradio" if sdk_choice == "gradio" else "streamlit"
try:
return importlib.metadata.version(pkg)
except importlib.metadata.PackageNotFoundError:
return "UNKNOWN"
def extract_code(text: str) -> str:
blocks = re.findall(r"```(?:\w*\n)?([\s\S]*?)```", text)
return blocks[-1].strip() if blocks else text.strip()
# — HF SPACE LOGGING —
def _get_space_jwt(repo_id: str):
url = f"{constants.ENDPOINT}/api/spaces/{repo_id}/jwt"
r = get_session().get(url, headers=build_hf_headers())
hf_raise_for_status(r)
return r.json()["token"]
def fetch_logs(repo_id: str, level: str) -> str:
jwt = _get_space_jwt(repo_id)
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 raw.startswith(b"data: "):
try:
ev = json.loads(raw[len(b"data: "):].decode())
ts = ev.get("timestamp","")
txt = ev.get("data","")
lines.append(f"[{ts}] {txt}")
except:
continue
return "\n".join(lines)
# — CORE LOOP —
def handle_user_message(
history,
sdk_choice: str,
gemini_api_key: str,
grounding_enabled: bool,
profile: gr.OAuthProfile | None,
oauth_token: gr.OAuthToken | None
):
if profile is None or oauth_token is None:
return history + [{"role":"assistant","content":"⚠️ Please log in first."}], "", "", "<p>No Space yet.</p>"
client = genai.Client(api_key=gemini_api_key)
system_msg = {
"role":"system",
"content":(
f"You are an AI assistant writing a HuggingFace Space using the "
f"{sdk_choice} SDK. After producing code, wait for logs; if errors appear, fix them."
)
}
chat = [system_msg] + history
code_fn = "app.py" if sdk_choice=="gradio" else "streamlit_app.py"
readme_fn = "README.md"
reqs_fn = "requirements.txt"
repo_id = f"{profile.username}/{profile.username}-auto-space"
build_logs = run_logs = ""
for _ in range(5):
tools = [Tool(google_search=GoogleSearch())] if grounding_enabled else []
cfg = GenerateContentConfig(tools=tools, response_modalities=["TEXT"])
resp = client.models.generate_content(
model="gemini-2.5-flash-preview-04-17",
contents=[m["content"] for m in chat],
config=cfg
)
raw = resp.text
code = extract_code(raw)
chat.append({"role":"assistant","content":code})
# write code
with open(code_fn, "w") as f:
f.write(code)
# write dynamic README
sdk_version = get_sdk_version(sdk_choice)
readme = f"""---
title: Wuhp Auto Space
emoji: 🐢
colorFrom: red
colorTo: pink
sdk: {sdk_choice}
sdk_version: {sdk_version}
app_file: {code_fn}
pinned: false
---
Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
"""
with open(readme_fn, "w") as f:
f.write(readme)
# write requirements
base_reqs = "pandas\n"
extra = "streamlit\n" if sdk_choice=="streamlit" else "gradio\n"
with open(reqs_fn, "w") as f:
f.write(base_reqs + extra)
# push to HF
create_repo(repo_id=repo_id, token=oauth_token.token,
exist_ok=True, repo_type="space", space_sdk=sdk_choice)
for fn in (code_fn, readme_fn, reqs_fn):
upload_file(path_or_fileobj=fn, path_in_repo=fn,
repo_id=repo_id, token=oauth_token.token,
repo_type="space")
build_logs = fetch_logs(repo_id, "build")
run_logs = fetch_logs(repo_id, "run")
if "ERROR" not in build_logs.upper() and "ERROR" not in run_logs.upper():
break
chat.append({
"role":"user",
"content":(
f"Build logs:\n{build_logs}\n\n"
f"Run logs:\n{run_logs}\n\n"
"Please fix the code."
)
})
time.sleep(2)
messages = [{"role":m["role"],"content":m["content"]} for m in chat if m["role"]!="system"]
iframe = f'<iframe src="https://huggingface.co/spaces/{repo_id}" width="100%" height="500px"></iframe>'
return messages, build_logs, run_logs, iframe
# — BUILD THE UI —
with gr.Blocks(title="HF Space Auto‑Builder") as demo:
gr.Markdown("## Sign in + Auto‑Build Spaces\n\n1. Sign in 2. Enter your prompt 3. Watch code, README, requirements, logs, and preview\n\n---")
# LOGIN & MODEL LISTING
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)
demo.load(list_private_models, inputs=None, outputs=models_md)
login_btn.click(show_profile, inputs=None, outputs=status_md)
login_btn.click(list_private_models, inputs=None, outputs=models_md)
# CONTROLS
sdk_choice = gr.Radio(["gradio","streamlit"], value="gradio", label="SDK template")
api_key = gr.Textbox(label="Gemini API Key", type="password")
grounding = gr.Checkbox(label="Enable grounding", value=False)
# CHAT + OUTPUTS
chatbot = gr.Chatbot(type="messages")
user_in = gr.Textbox(placeholder="Your prompt…", label="Prompt")
send_btn = gr.Button("Send")
build_box = gr.Textbox(label="Build logs", lines=5, interactive=False)
run_box = gr.Textbox(label="Run logs", lines=5, interactive=False)
preview = gr.HTML("<p>No Space yet.</p>")
send_btn.click(
fn=handle_user_message,
inputs=[chatbot, sdk_choice, api_key, grounding],
outputs=[chatbot, build_box, run_box, preview]
)
# — Refresh Logs button —
def _refresh(profile: gr.OAuthProfile | None, oauth_token: gr.OAuthToken | None):
if not profile or not oauth_token:
return "", ""
repo = f"{profile.username}/{profile.username}-auto-space"
return fetch_logs(repo, "build"), fetch_logs(repo, "run")
refresh_btn = gr.Button("Refresh Logs")
# Gradio will auto‑inject `profile` and `oauth_token` here.
refresh_btn.click(_refresh, inputs=None, outputs=[build_box, run_box])
demo.launch(server_name="0.0.0.0", server_port=7860)