""" 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 "

暂无公开反馈

" blocks = [] for item in items: created = (item.get("created") or "")[:19] username = item.get("username") or "匿名" text = (item.get("text") or "").replace("\n", "
") img_note = "" if item.get("images"): img_note = f"
附带 {len(item['images'])} 张图片
" blocks.append( f"
" f"
{username} " f"{created}
" f"
{text}
{img_note}
" ) 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("""

🛠️ 3D 模型优化服务

上传 3D 模型，自动优化处理，完成后下载
GLBGLTF FBXOBJ →GLB 输出

⚠️ 测试阶段，请优先上传 GLB 格式文件

""") 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("

Powered by Blender · Worker 每30秒检查新任务

") 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("""

🛠️ 3D 模型优化服务

上传 3D 模型，自动优化处理，完成后下载
GLB GLTF FBX OBJ → GLB 输出
⚠️ 测试阶段，请优先上传 GLB 格式文件
""") 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("""
Powered by Blender · Worker 每30秒检查新任务
""") if __name__ == "__main__": log("Gradio 前端启动") demo.launch()