#!/usr/bin/env python3 # -*- coding: utf-8 -*- """DSH 机视频生成流水线 Web 界面（端口 8898）—— 三步式。步骤 1：上传参考图 → 点「👁️ 识图」→ 显示模型对图片的识别结果步骤 2：根据识图结果填写描述 → 点「📝 生成提示词」→ 显示六段式 Ref2VA 提示词步骤 3：确认提示词 → 点「🎬 生成视频」→ 提交 GPU 机 V18 工作流 → 视频回传下载中心运行：vlm-env/bin/python web_app.py """ import json import sys import time import uuid import threading from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from v18_pipeline import run_pipeline, describe_image, build_prompt_from_vision # noqa: E402 from flask import Flask, request, jsonify, render_template_string, send_from_directory APP_DIR = Path(__file__).resolve().parent UPLOAD_DIR = APP_DIR / "uploads" UPLOAD_DIR.mkdir(exist_ok=True) # 任务状态存储 TASKS = {} TASKS_LOCK = threading.Lock() # 步骤1/2 的异步任务 STEPS = {} STEPS_LOCK = threading.Lock() app = Flask(__name__) app.config["MAX_CONTENT_LENGTH"] = 64 * 1024 * 1024 # 64MB HTML = """ DSH 视频生成流水线（V18 Ref2VA · 三步式）

🎬 DSH 视频生成流水线（三步式）

① 识图 → ② 生成提示词 → ③ 生成视频。每步独立按钮，结果直接显示在各步骤下方区域。

1

上传参考图 + 识图

参考图片（第 1 张 = <Picture 1>，可多选，上传顺序即编号）

👁️ 识图结果等待识图

点击上方「👁️ 识图」按钮，识别结果会显示在这里。

2

填写描述 + 生成提示词

描述（自己填写：画面内容、动作、运镜、台词等。识图结果仅供参考，不会自动填入）

📝 提示词结果等待生成

点击上方「📝 生成提示词」按钮，六段式提示词会显示在这里。

生成的提示词（可手动修改后再生成视频）

3

生成视频

时长（秒）

分辨率

步数

Seed（0=随机）

帧率

🎬 视频结果等待生成

点击上方「🎬 生成视频」按钮，提交 GPU 机 V18 工作流，完成后视频显示在这里。

提示：GPU 机队列忙时需等待；单次生成约 2–15 分钟（视排队）。若未生成提示词，将自动走完整流程。

📋 最近任务

加载中...

""" def get_resolution(res_str: str): w, h = res_str.lower().split("x") return int(w), int(h) def save_uploads(files, subdir: str): """保存上传文件，返回本地路径列表。""" job_dir = UPLOAD_DIR / subdir job_dir.mkdir(exist_ok=True) saved = [] for i, f in enumerate(files): ext = Path(f.filename).suffix or ".jpg" p = job_dir / f"pic_{i+1}{ext}" f.save(p) saved.append(str(p)) return saved @app.route("/") def index(): return render_template_string(HTML) # ---------------- 步骤 1：识图 ---------------- @app.route("/api/vision", methods=["POST"]) def vision(): files = request.files.getlist("images") if not files: return jsonify({"error": "请先选择参考图片"}), 400 job_id = uuid.uuid4().hex[:10] saved = save_uploads(files, f"vision_{job_id}") with STEPS_LOCK: STEPS[job_id] = { "job_id": job_id, "kind": "vision", "images": saved, "status": "running", "log": "正在识图...", "result": None, "created": time.time() * 1000, "error": None, } threading.Thread(target=vision_worker, args=(job_id,), daemon=True).start() return jsonify({"job_id": job_id}) def vision_worker(job_id: str): def set_job(**kw): with STEPS_LOCK: if job_id in STEPS: STEPS[job_id].update(kw) try: j = STEPS[job_id] set_job(log="正在识图（本地 Qwen2.5-VL-3B，约 1–2 分钟）...") vision = describe_image(j["images"][0]) set_job(status="done", log="✅ 识图完成", result=vision) except Exception as e: import traceback traceback.print_exc() set_job(status="error", error=str(e), log=f"❌ {e}") # ---------------- 步骤 2：生成提示词 ---------------- @app.route("/api/build-prompt", methods=["POST"]) def build_prompt(): files = request.files.getlist("images") requirement = (request.form.get("requirement") or "").strip() if not files and not requirement: return jsonify({"error": "请选择参考图片或填写描述"}), 400 job_id = uuid.uuid4().hex[:10] saved = save_uploads(files, f"prompt_{job_id}") vision = (request.form.get("vision") or "").strip() with STEPS_LOCK: STEPS[job_id] = { "job_id": job_id, "kind": "prompt", "images": saved, "requirement": requirement, "vision": vision, "status": "running", "log": "正在生成提示词...", "result": None, "created": time.time() * 1000, "error": None, } threading.Thread(target=prompt_worker, args=(job_id,), daemon=True).start() return jsonify({"job_id": job_id}) def prompt_worker(job_id: str): def set_job(**kw): with STEPS_LOCK: if job_id in STEPS: STEPS[job_id].update(kw) try: j = STEPS[job_id] set_job(log="正在生成六段式提示词（约 1–3 分钟）...") if j["images"]: prompt = build_prompt_from_vision(j["images"][0], j["vision"], j["requirement"], 10) else: # 无图：直接让模型按要求生成提示词 prompt = build_prompt_from_vision(None, j["vision"], j["requirement"], 10) set_job(status="done", log="✅ 提示词生成完成", result=prompt) except Exception as e: import traceback traceback.print_exc() set_job(status="error", error=str(e), log=f"❌ {e}") # ---------------- 步骤 3：生成视频 ---------------- @app.route("/api/submit", methods=["POST"]) def submit(): files = request.files.getlist("images") prompt_override = (request.form.get("prompt_override") or "").strip() vision_override = (request.form.get("vision_override") or "").strip() requirement = (request.form.get("requirement") or "").strip() if not files: return jsonify({"error": "请选择参考图片"}), 400 if not prompt_override and not requirement: return jsonify({"error": "请先在第 2 步生成提示词，或填写描述"}), 400 task_id = uuid.uuid4().hex[:10] saved = save_uploads(files, task_id) duration = int(request.form.get("duration", 10)) w, h = get_resolution(request.form.get("resolution", "736x1280")) steps = int(request.form.get("steps", 8)) seed = int(request.form.get("seed", 666666)) fps = float(request.form.get("fps", 24)) if seed == 0: import random seed = random.randint(1, 2**31) with TASKS_LOCK: TASKS[task_id] = { "task_id": task_id, "images": saved, "requirement": requirement, "params": {"width": w, "height": h, "duration": duration, "steps": steps, "seed": seed, "fps": fps}, "prompt_override": prompt_override, "vision_override": vision_override, "status": "queued", "created": time.time() * 1000, "log": "排队中...", "prompt": None, "vision_description": None, "video_url": None, "error": None, } threading.Thread(target=worker, args=(task_id,), daemon=True).start() return jsonify({"task_id": task_id}) def worker(task_id: str): def set_task(**kw): with TASKS_LOCK: if task_id in TASKS: TASKS[task_id].update(kw) try: t = TASKS[task_id] if t["prompt_override"]: set_task(status="running", log="使用你的提示词，提交 GPU 机（约 2–15 分钟）...") else: set_task(status="running", log="正在解析图片（识图阶段，约 1–2 分钟）...") result = run_pipeline( t["images"][0], t["requirement"], width=t["params"]["width"], height=t["params"]["height"], duration=t["params"]["duration"], seed=t["params"]["seed"], steps=t["params"]["steps"], fps=t["params"]["fps"], prompt_override=t["prompt_override"] or None, vision_override=t["vision_override"] or None, on_vision=lambda v: set_task( vision_description=v, log="✅ 识图完成，正在生成提示词并提交 GPU（约 5–15 分钟）..."), ) set_task(status="done", log="✅ 完成", prompt=result["prompt"], vision_description=result.get("vision_description", ""), video_url=result["video_url"]) except Exception as e: import traceback traceback.print_exc() set_task(status="error", error=str(e), log=f"❌ {e}") @app.route("/api/task/") def task_status(task_id): with TASKS_LOCK: t = TASKS.get(task_id) if not t: return jsonify({"error": "not found"}), 404 return jsonify(t) @app.route("/api/tasks") def task_list(): with TASKS_LOCK: items = sorted(TASKS.values(), key=lambda x: x["created"], reverse=True)[:20] return jsonify(items) @app.route("/api/steps/") def step_status(job_id): with STEPS_LOCK: j = STEPS.get(job_id) if not j: return jsonify({"error": "not found"}), 404 return jsonify(j) @app.route("/uploads/") def uploaded_file(filename): return send_from_directory(str(UPLOAD_DIR), filename) if __name__ == "__main__": print("DSH 视频流水线 Web（三步式）: http://192.168.31.76:8898") app.run(host="0.0.0.0", port=8898, debug=False, threaded=True)