model-optimizer / app_hf.py
wangyiyi666's picture
Upload folder using huggingface_hub
a030c8b verified
Raw
History Blame Contribute Delete
24.5 kB
"""
3D 模型优化服务 - HF Space 版本
Gradio + HF Hub API | 含用户建议反馈
"""
import os
import json
import uuid
from datetime import datetime
import gradio as gr
from huggingface_hub import HfApi, list_repo_files
from feedback_util import fetch_public_feedback, submit_feedback
# === 配置 ===
REPO_ID = "wangyiyi666/model-optimizer-queue"
REPO_TYPE = "dataset"
HF_TOKEN = os.environ.get("HF_TOKEN", "")
SUPPORTED_FORMATS = [".glb", ".gltf", ".fbx", ".obj"]
api = HfApi(token=HF_TOKEN)
def log(msg):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] [SPACE] {msg}")
def upload_model(file, progress=gr.Progress()):
"""处理用户上传的模型文件"""
if file is None:
return "### ⚠️ 请先选择文件", ""
filename = os.path.basename(file.name if hasattr(file, "name") else file)
ext = os.path.splitext(filename)[1].lower()
log(f"用户上传: {filename}")
if ext not in SUPPORTED_FORMATS:
log(f"格式不支持: {ext}")
return f"### ❌ 不支持的格式: `{ext}`\n\n支持的格式: {', '.join(SUPPORTED_FORMATS)}", ""
task_id = uuid.uuid4().hex
target_name = f"{task_id}{ext}"
try:
progress(0.3, desc="正在上传模型文件...")
file_path = file.name if hasattr(file, "name") else file
api.upload_file(
path_or_fileobj=file_path,
path_in_repo=f"inbox/{target_name}",
repo_id=REPO_ID,
repo_type=REPO_TYPE,
)
progress(0.7, desc="正在创建任务...")
name_no_ext = os.path.splitext(filename)[0]
meta = json.dumps({
"task_id": task_id,
"filename": filename,
"name_no_ext": name_no_ext,
"status": "pending",
"created": str(datetime.now()),
})
api.upload_file(
path_or_fileobj=meta.encode(),
path_in_repo=f"inbox/{task_id}.json",
repo_id=REPO_ID,
repo_type=REPO_TYPE,
)
progress(1.0, desc="上传完成!")
log(f"上传成功: {target_name}, 任务ID: {task_id}")
file_size = os.path.getsize(file_path)
size_str = f"{file_size / 1024:.1f} KB" if file_size < 1024 * 1024 else f"{file_size / 1024 / 1024:.1f} MB"
return (
f"### ✅ **上传成功!**\n\n"
f"| 项目 | 信息 |\n"
f"|------|------|\n"
f"| 📋 任务ID | `{task_id}` |\n"
f"| 📁 文件名 | {filename} |\n"
f"| 📦 文件大小 | {size_str} |\n"
f"| ⏱️ 状态 | 等待优化处理 |\n\n"
f"> ⚠️ **务必保存好任务ID,这是您下载优化结果的唯一凭证!**"
), task_id
except Exception as e:
log(f"上传失败: {e}")
return f"### ❌ 上传失败\n\n```\n{str(e)}\n```", ""
def check_status(task_id):
"""查询任务状态"""
if not task_id or len(task_id.strip()) == 0:
return "### ⚠️ 请输入任务ID", gr.update(visible=False)
task_id = task_id.strip()
log(f"查询: {task_id}")
try:
files = list_repo_files(REPO_ID, repo_type=REPO_TYPE, token=HF_TOKEN)
outbox_matches = [f for f in files if f.startswith(f"outbox/{task_id}") and not f.endswith(".json")]
if outbox_matches:
result_path = outbox_matches[0]
ext = os.path.splitext(result_path)[1].upper().lstrip(".")
log(f"任务 {task_id} 已完成")
return (
f"### ✅ 优化完成!\n\n"
f"| 项目 | 信息 |\n"
f"|------|------|\n"
f"| 📋 任务ID | `{task_id}` |\n"
f"| 📄 格式 | {ext} |\n\n"
f"> 👇 点击下方「下载模型」按钮获取优化后的文件"
), gr.update(visible=True)
inbox_files = [f for f in files if f.startswith(f"inbox/{task_id}")]
if inbox_files:
log(f"任务 {task_id} 仍在队列中")
return (
f"### ⏳ 处理中\n\n"
f"任务 `{task_id}` 正在优化队列中,请稍后再查询。\n\n"
f"> Worker 每 **30秒** 检查一次新任务,优化完成后即可下载。"
), gr.update(visible=False)
log(f"任务 {task_id} 未找到")
return f"### ❌ 未找到任务\n\n任务ID `{task_id}` 不存在,请检查是否输入正确。", gr.update(visible=False)
except Exception as e:
log(f"查询出错: {e}")
return f"### ❌ 查询出错\n\n```\n{str(e)}\n```", gr.update(visible=False)
def download_model(task_id):
"""下载模型并记录取件事件"""
if not task_id or len(task_id.strip()) == 0:
return None, "### ⚠️ 请先查询任务ID"
task_id = task_id.strip()
log(f"用户下载: {task_id}")
try:
files = list_repo_files(REPO_ID, repo_type=REPO_TYPE, token=HF_TOKEN)
outbox_matches = [f for f in files if f.startswith(f"outbox/{task_id}") and not f.endswith(".json")]
if not outbox_matches:
return None, "### ❌ 未找到优化结果"
result_path = outbox_matches[0]
download_url = f"https://huggingface.co/datasets/{REPO_ID}/resolve/main/{result_path}"
# 记录下载事件到 HF Dataset
from datetime import datetime
record = json.dumps({
"task_id": task_id,
"downloaded_at": datetime.now().isoformat(),
"result_file": result_path,
})
try:
api.upload_file(
path_or_fileobj=record.encode(),
path_in_repo=f"outbox/{task_id}.downloaded.json",
repo_id=REPO_ID,
repo_type=REPO_TYPE,
)
log(f"已记录下载: {task_id}")
except Exception as e:
log(f"记录下载失败(不影响下载): {e}")
return (
None,
f"### ✅ 下载链接已生成\n\n"
f"> 👇 [点击此处下载优化后的模型]({download_url})\n\n"
f"*下载记录已通知管理端*"
)
except Exception as e:
log(f"下载出错: {e}")
return None, f"### ❌ 下载出错\n\n```\n{str(e)}\n```"
def delete_task(task_id):
"""删除任务(清理 inbox + outbox 文件)"""
if not task_id or len(task_id.strip()) == 0:
return "### ⚠️ 请输入任务ID"
task_id = task_id.strip()
log(f"删除任务: {task_id}")
try:
files = list_repo_files(REPO_ID, repo_type=REPO_TYPE, token=HF_TOKEN)
to_delete = [f for f in files if task_id in f and f != ".gitattributes"]
if not to_delete:
return f"### ❌ 未找到任务\n\n任务ID `{task_id}` 不存在。"
deleted = []
for filepath in to_delete:
try:
api.delete_file(filepath, REPO_ID, repo_type=REPO_TYPE)
deleted.append(filepath)
log(f"已删除: {filepath}")
except Exception as e:
log(f"删除失败 {filepath}: {e}")
file_list = "\n".join([f"- `{f}`" for f in deleted])
return f"### 🗑️ 删除成功\n\n已删除 **{len(deleted)}** 个文件:\n\n{file_list}"
except Exception as e:
log(f"删除出错: {e}")
return f"### ❌ 删除出错\n\n```\n{str(e)}\n```"
def render_public_feedback():
items = fetch_public_feedback()
if not items:
return "<p style='color:#888;'>暂无公开反馈</p>"
blocks = []
for item in items:
created = (item.get("created") or "")[:19]
username = item.get("username") or "匿名"
text = (item.get("text") or "").replace("\n", "<br>")
img_note = ""
if item.get("images"):
img_note = f"<div style='color:#666;font-size:13px;margin-top:6px;'>附带 {len(item['images'])} 张图片</div>"
blocks.append(
f"<div style='border:1px solid #eee;border-radius:8px;padding:14px;margin-bottom:12px;background:#fafafa;'>"
f"<div style='font-weight:600;'>{username} "
f"<span style='color:#999;font-size:12px;'>{created}</span></div>"
f"<div style='margin-top:8px;line-height:1.6;'>{text}</div>{img_note}</div>"
)
return "".join(blocks)
def handle_feedback_submit(username, text, images, is_public):
try:
image_paths = []
if images:
if isinstance(images, list):
image_paths = [item.name if hasattr(item, "name") else item for item in images]
else:
image_paths = [images.name if hasattr(images, "name") else images]
meta = submit_feedback(username, text, image_paths, is_public)
visibility = "已公开" if meta.get("is_public") else "仅管理员可见"
return f"### ✅ 反馈提交成功\n\n| 项目 | 信息 |\n|------|------|\n| 反馈ID | `{meta['feedback_id']}` |\n| 展示范围 | {visibility} |"
except Exception as e:
log(f"反馈提交失败: {e}")
return f"### ❌ 提交失败\n\n```\n{str(e)}\n```"
custom_css = """
.main-title { text-align: center; margin-bottom: 0.5em; }
.sub-title { text-align: center; color: #666; font-size: 1.1em; margin-bottom: 1.5em; }
.format-badge {
display: inline-block; background: #e3f2fd; color: #1565c0;
padding: 4px 12px; border-radius: 16px; margin: 2px; font-size: 0.9em; font-weight: 500;
}
footer { display: none !important; }
"""
with gr.Blocks(
title="3D 模型优化服务",
theme=gr.themes.Soft(primary_hue="blue", secondary_hue="slate"),
css=custom_css,
) as demo:
gr.HTML("""
<div class="main-title"><h1>🛠️ 3D 模型优化服务</h1></div>
<div class="sub-title">上传 3D 模型,自动优化处理,完成后下载</div>
<div style="text-align:center;margin-bottom:1.5em;">
<span class="format-badge">GLB</span><span class="format-badge">GLTF</span>
<span class="format-badge">FBX</span><span class="format-badge">OBJ</span>
<span style="margin:0 8px;">→</span><span class="format-badge">GLB 输出</span>
</div>
<p style="text-align:center;color:#e65100;">⚠️ 测试阶段,请优先上传 GLB 格式文件</p>
""")
with gr.Tabs():
with gr.Tab("📤 上传模型", id="upload"):
with gr.Row(equal_height=True):
with gr.Column(scale=1):
file_input = gr.File(
label="选择 3D 模型文件",
file_types=[".glb", ".gltf", ".fbx", ".obj"],
type="filepath",
height=200,
)
upload_btn = gr.Button("🚀 提交优化", variant="primary", size="lg")
with gr.Column(scale=1):
upload_result = gr.Markdown(
value="### 📋 等待上传\n\n选择文件后点击「提交优化」按钮。",
label="处理结果",
)
task_id_output = gr.Textbox(label="📋 任务ID(复制保存)", interactive=False)
upload_btn.click(upload_model, inputs=[file_input], outputs=[upload_result, task_id_output])
with gr.Tab("🔍 查询结果", id="query"):
with gr.Row(equal_height=True):
with gr.Column(scale=1):
task_id_input = gr.Textbox(label="输入任务ID", placeholder="粘贴完整任务ID", max_lines=1)
with gr.Row():
check_btn = gr.Button("🔍 查询状态", variant="primary", size="lg")
delete_btn = gr.Button("🗑️ 删除任务", variant="stop", size="lg")
download_btn = gr.Button("📥 下载模型", variant="secondary", size="lg", visible=False)
with gr.Column(scale=1):
status_result = gr.Markdown(
value="### 📋 等待查询\n\n输入任务ID后点击「查询状态」按钮。",
label="任务状态",
)
check_btn.click(check_status, inputs=[task_id_input], outputs=[status_result, download_btn])
delete_btn.click(delete_task, inputs=[task_id_input], outputs=[status_result])
download_btn.click(download_model, inputs=[task_id_input], outputs=[download_btn, status_result])
with gr.Tab("💬 提交反馈", id="feedback"):
gr.Markdown("欢迎提交使用建议或问题反馈,可附带截图。管理员会在本地管理面板查看全部反馈。")
feedback_username = gr.Textbox(label="用户名 / 昵称", placeholder="可选,默认匿名")
feedback_text = gr.Textbox(label="反馈内容", lines=6, placeholder="请描述您的建议或遇到的问题...")
feedback_images = gr.File(
label="截图 / 图片(可选,可多选)",
file_count="multiple",
file_types=["image"],
type="filepath",
)
feedback_public = gr.Checkbox(
label="允许对外公开展示(勾选后其他用户可在「公开反馈」页看到文字内容)",
value=False,
)
feedback_submit_btn = gr.Button("提交反馈", variant="primary")
feedback_result = gr.Markdown()
feedback_submit_btn.click(
handle_feedback_submit,
inputs=[feedback_username, feedback_text, feedback_images, feedback_public],
outputs=[feedback_result],
)
with gr.Tab("📣 公开反馈", id="public_feedback"):
refresh_public_btn = gr.Button("🔄 刷新公开反馈")
public_feedback_html = gr.HTML(value=render_public_feedback())
refresh_public_btn.click(lambda: render_public_feedback(), outputs=[public_feedback_html])
gr.HTML("<p style='text-align:center;color:#999;margin-top:1em;'>Powered by Blender · Worker 每30秒检查新任务</p>")
if __name__ == "__main__":
log("Gradio 前端启动")
demo.launch()
"""
3D 模型优化服务 - HF Space 版本
Gradio + HF Hub API | 美观 UI
"""
import gradio as gr
from huggingface_hub import HfApi, list_repo_files
import os
import json
import uuid
from datetime import datetime
# === 配置 ===
REPO_ID = "wangyiyi666/model-optimizer-queue"
REPO_TYPE = "dataset"
HF_TOKEN = os.environ.get("HF_TOKEN", "")
SUPPORTED_FORMATS = [".glb", ".gltf", ".fbx", ".obj"]
api = HfApi(token=HF_TOKEN)
def log(msg):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] [SPACE] {msg}")
def upload_model(file, progress=gr.Progress()):
"""处理用户上传的模型文件"""
if file is None:
return "### ⚠️ 请先选择文件", ""
filename = os.path.basename(file.name if hasattr(file, 'name') else file)
ext = os.path.splitext(filename)[1].lower()
log(f"用户上传: {filename}")
if ext not in SUPPORTED_FORMATS:
log(f"格式不支持: {ext}")
return f"### ❌ 不支持的格式: `{ext}`\n\n支持的格式: {', '.join(SUPPORTED_FORMATS)}", ""
task_id = uuid.uuid4().hex
target_name = f"{task_id}{ext}"
try:
progress(0.3, desc="正在上传模型文件...")
file_path = file.name if hasattr(file, 'name') else file
api.upload_file(
path_or_fileobj=file_path,
path_in_repo=f"inbox/{target_name}",
repo_id=REPO_ID,
repo_type=REPO_TYPE
)
progress(0.7, desc="正在创建任务...")
name_no_ext = os.path.splitext(filename)[0]
meta = json.dumps({
"task_id": task_id,
"filename": filename,
"name_no_ext": name_no_ext,
"status": "pending",
"created": str(datetime.now())
})
api.upload_file(
path_or_fileobj=meta.encode(),
path_in_repo=f"inbox/{task_id}.json",
repo_id=REPO_ID,
repo_type=REPO_TYPE
)
progress(1.0, desc="上传完成!")
log(f"上传成功: {target_name}, 任务ID: {task_id}")
file_size = os.path.getsize(file_path)
size_str = f"{file_size / 1024:.1f} KB" if file_size < 1024 * 1024 else f"{file_size / 1024 / 1024:.1f} MB"
return (
f"### ✅ **上传成功!**\n\n"
f"| 项目 | 信息 |\n"
f"|------|------|\n"
f"| 📋 任务ID | `{task_id}` |\n"
f"| 📁 文件名 | {filename} |\n"
f"| 📦 文件大小 | {size_str} |\n"
f"| ⏱️ 状态 | 等待优化处理 |\n\n"
f"> ⚠️ **务必保存好任务ID,这是您下载优化结果的唯一凭证!**"
), task_id
except Exception as e:
log(f"上传失败: {e}")
return f"### ❌ 上传失败\n\n```\n{str(e)}\n```", ""
def check_status(task_id):
"""查询任务状态"""
if not task_id or len(task_id.strip()) == 0:
return "### ⚠️ 请输入任务ID"
task_id = task_id.strip()
log(f"查询: {task_id}")
try:
files = list_repo_files(REPO_ID, repo_type=REPO_TYPE, token=HF_TOKEN)
# 检查 outbox 是否有结果: outbox/{task_id}.xxx(保持原始格式)
outbox_matches = [f for f in files if f.startswith(f"outbox/{task_id}") and not f.endswith('.json')]
if outbox_matches:
result_path = outbox_matches[0]
ext = os.path.splitext(result_path)[1].upper().lstrip('.')
download_url = f"https://huggingface.co/datasets/{REPO_ID}/resolve/main/{result_path}"
log(f"任务 {task_id} 已完成")
return (
f"### ✅ 优化完成!\n\n"
f"| 项目 | 信息 |\n"
f"|------|------|\n"
f"| 📋 任务ID | `{task_id}` |\n"
f"| 📄 格式 | {ext} |\n\n"
f"> 👇 [点击此处下载优化后的模型]({download_url})"
)
# 检查 inbox 是否还在排队
inbox_files = [f for f in files if f.startswith(f"inbox/{task_id}")]
if inbox_files:
log(f"任务 {task_id} 仍在队列中")
return (
f"### ⏳ 处理中\n\n"
f"任务 `{task_id}` 正在优化队列中,请稍后再查询。\n\n"
f"> Worker 每 **30秒** 检查一次新任务,优化完成后即可下载。"
)
log(f"任务 {task_id} 未找到")
return (
f"### ❌ 未找到任务\n\n"
f"任务ID `{task_id}` 不存在,请检查是否输入正确。"
)
except Exception as e:
log(f"查询出错: {e}")
return f"### ❌ 查询出错\n\n```\n{str(e)}\n```"
def delete_task(task_id):
"""删除任务(清理 inbox + outbox 文件)"""
if not task_id or len(task_id.strip()) == 0:
return "### ⚠️ 请输入任务ID"
task_id = task_id.strip()
log(f"删除任务: {task_id}")
try:
files = list_repo_files(REPO_ID, repo_type=REPO_TYPE, token=HF_TOKEN)
to_delete = [f for f in files if task_id in f and f != '.gitattributes']
if not to_delete:
return f"### ❌ 未找到任务\n\n任务ID `{task_id}` 不存在。"
deleted = []
for filepath in to_delete:
try:
api.delete_file(filepath, REPO_ID, repo_type=REPO_TYPE)
deleted.append(filepath)
log(f"已删除: {filepath}")
except Exception as e:
log(f"删除失败 {filepath}: {e}")
file_list = "\n".join([f"- `{f}`" for f in deleted])
return f"### 🗑️ 删除成功\n\n已删除 **{len(deleted)}** 个文件:\n\n{file_list}"
except Exception as e:
log(f"删除出错: {e}")
return f"### ❌ 删除出错\n\n```\n{str(e)}\n```"
# === 自定义 CSS ===
custom_css = """
.main-title {
text-align: center;
margin-bottom: 0.5em;
}
.sub-title {
text-align: center;
color: #666;
font-size: 1.1em;
margin-bottom: 1.5em;
}
.format-badge {
display: inline-block;
background: #e3f2fd;
color: #1565c0;
padding: 4px 12px;
border-radius: 16px;
margin: 2px;
font-size: 0.9em;
font-weight: 500;
}
footer { display: none !important; }
"""
# === Gradio 界面 ===
with gr.Blocks(
title="3D 模型优化服务",
theme=gr.themes.Soft(primary_hue="blue", secondary_hue="slate"),
css=custom_css
) as demo:
gr.HTML("""
<div class="main-title">
<h1>🛠️ 3D 模型优化服务</h1>
</div>
<div class="sub-title">
上传 3D 模型,自动优化处理,完成后下载
</div>
<div style="text-align: center; margin-bottom: 20px;">
<span class="format-badge">GLB</span>
<span class="format-badge">GLTF</span>
<span class="format-badge">FBX</span>
<span class="format-badge">OBJ</span>
<span style="margin: 0 8px; color: #999;">→</span>
<span class="format-badge" style="background: #e8f5e9; color: #2e7d32;">GLB 输出</span>
</div>
<div style="text-align: center; margin-bottom: 16px; padding: 8px 16px; background: #fff3cd; border-radius: 8px; color: #856404; font-size: 14px;">
⚠️ 测试阶段,请优先上传 GLB 格式文件
</div>
""")
with gr.Tabs():
with gr.Tab("📤 上传模型", id="upload"):
with gr.Row(equal_height=True):
with gr.Column(scale=1):
file_input = gr.File(
label="选择 3D 模型文件",
file_types=[".glb", ".gltf", ".fbx", ".obj"],
type="filepath",
height=200,
)
upload_btn = gr.Button(
"🚀 提交优化",
variant="primary",
size="lg",
)
with gr.Column(scale=1):
upload_result = gr.Markdown(
value="### 📋 等待上传\n\n选择文件后点击「提交优化」按钮。",
label="处理结果",
)
task_id_output = gr.Textbox(
label="📋 任务ID(复制保存)",
interactive=False,
)
upload_btn.click(
upload_model,
inputs=[file_input],
outputs=[upload_result, task_id_output]
)
with gr.Tab("🔍 查询结果", id="query"):
with gr.Row(equal_height=True):
with gr.Column(scale=1):
task_id_input = gr.Textbox(
label="输入任务ID",
placeholder="粘贴完整任务ID",
max_lines=1,
)
with gr.Row():
check_btn = gr.Button(
"🔍 查询状态",
variant="primary",
size="lg",
)
delete_btn = gr.Button(
"🗑️ 删除任务",
variant="stop",
size="lg",
)
with gr.Column(scale=1):
status_result = gr.Markdown(
value="### 📋 等待查询\n\n输入任务ID后点击「查询状态」按钮。",
label="任务状态",
)
check_btn.click(
check_status,
inputs=[task_id_input],
outputs=[status_result]
)
delete_btn.click(
delete_task,
inputs=[task_id_input],
outputs=[status_result]
)
gr.HTML("""
<div style="text-align: center; margin-top: 20px; padding: 15px; border-top: 1px solid #eee; color: #999; font-size: 0.85em;">
Powered by Blender · Worker 每30秒检查新任务
</div>
""")
if __name__ == "__main__":
log("Gradio 前端启动")
demo.launch()