"""
模型优化服务 - Flask 前端
第一版:本地测试模式(先跑通完整通路)
后续部署到 HF Space 时可切换为 Gradio 版本
"""
from flask import Flask, request, render_template_string, send_file, jsonify
import os
import uuid
import shutil
from datetime import datetime
app = Flask(__name__)
# === 配置 ===
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
INBOX = os.path.join(BASE_DIR, "..", "local", "tmp", "inbox")
OUTBOX = os.path.join(BASE_DIR, "..", "local", "tmp", "outbox")
os.makedirs(INBOX, exist_ok=True)
os.makedirs(OUTBOX, exist_ok=True)
SUPPORTED_FORMATS = [".glb", ".gltf", ".fbx", ".obj"]
def log(msg):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"[{timestamp}] [WEB] {msg}")
HTML_TEMPLATE = """
3D 模型优化服务
🛠️ 3D 模型优化服务
📤 上传模型
{% if upload_msg %}
{{ upload_msg }}
{% endif %}
🔍 查询结果
{% if status_msg %}
{{ status_msg }}
{% endif %}
{% if download_ready %}
⬇️ 下载优化结果
{% endif %}
输出格式: GLB | Worker 每30秒检查一次任务
"""
@app.route("/")
def index():
return render_template_string(HTML_TEMPLATE)
@app.route("/upload", methods=["POST"])
def upload():
file = request.files.get("model")
if not file or file.filename == "":
return render_template_string(HTML_TEMPLATE,
upload_msg="请选择文件", upload_class="error")
filename = file.filename
ext = os.path.splitext(filename)[1].lower()
log(f"收到上传: {filename}")
if ext not in SUPPORTED_FORMATS:
log(f"格式不支持: {ext}")
return render_template_string(HTML_TEMPLATE,
upload_msg=f"不支持的格式: {ext}", upload_class="error")
# 生成任务ID并保存文件
task_id = str(uuid.uuid4())[:8]
target_path = os.path.join(INBOX, f"{task_id}{ext}")
file.save(target_path)
file_size = os.path.getsize(target_path)
log(f"文件已保存: {task_id}{ext} ({file_size} bytes), 任务ID: {task_id}")
return render_template_string(HTML_TEMPLATE,
upload_msg=f"✅ 上传成功!任务ID: {task_id} (请保存此ID用于查询结果)",
upload_class="success")
@app.route("/status")
def status():
task_id = request.args.get("task_id", "").strip()
if not task_id:
return render_template_string(HTML_TEMPLATE,
status_msg="请输入任务ID", status_class="error", query_id=task_id)
log(f"查询状态: {task_id}")
# 检查 outbox
result_file = os.path.join(OUTBOX, f"{task_id}_optimized.glb")
if os.path.exists(result_file):
size = os.path.getsize(result_file)
log(f"任务 {task_id} 已完成 ({size} bytes)")
return render_template_string(HTML_TEMPLATE,
status_msg=f"✅ 优化完成!文件大小: {size} bytes",
status_class="success", download_ready=True,
download_id=task_id, query_id=task_id)
# 检查 inbox
for f in os.listdir(INBOX):
if f.startswith(task_id):
log(f"任务 {task_id} 仍在队列中")
return render_template_string(HTML_TEMPLATE,
status_msg="⏳ 处理中,请稍后再查询...",
status_class="waiting", query_id=task_id)
log(f"任务 {task_id} 未找到")
return render_template_string(HTML_TEMPLATE,
status_msg="❌ 未找到该任务,请检查ID是否正确",
status_class="error", query_id=task_id)
@app.route("/download/")
def download(task_id):
result_file = os.path.join(OUTBOX, f"{task_id}_optimized.glb")
if os.path.exists(result_file):
log(f"用户下载: {task_id}_optimized.glb")
return send_file(result_file, as_attachment=True,
download_name=f"{task_id}_optimized.glb")
return "文件不存在", 404
if __name__ == "__main__":
log("=" * 40)
log("Web 前端启动")
log(f" Inbox: {INBOX}")
log(f" Outbox: {OUTBOX}")
log("=" * 40)
app.run(host="0.0.0.0", port=5000, debug=True)