From a095bdd927084ed266744cdf9360898c670bcbe3 Mon Sep 17 00:00:00 2001 From: chenjw28 <792430652@qq.com> Date: Mon, 15 Jun 2026 13:52:20 +0800 Subject: [PATCH] x --- docker/Dockerfile | 4 +- docker/docker-compose.yml | 8 +- docker/scripts/gradio_client.py | 275 ++++++++++++++++++++++++++++++++ 3 files changed, 280 insertions(+), 7 deletions(-) create mode 100644 docker/scripts/gradio_client.py diff --git a/docker/Dockerfile b/docker/Dockerfile index 5b0d86e..2010f19 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -303,8 +303,8 @@ RUN set -ex && \ # =========================================================================== RUN set -ex && \ export http_proxy=${GIT_PROXY} https_proxy=${GIT_PROXY} && \ - ${VENV}/bin/pip install 'mineru[core]' click gradio && \ - ${VENV}/bin/python -c "import torch; import vllm; import mineru; import click; import gradio; import mineru.cli.gradio_app; print('MinerU + Gradio CLI imports OK')" && \ + ${VENV}/bin/pip install 'mineru[core]' click gradio httpx && \ + ${VENV}/bin/python -c "import torch; import vllm; import mineru; import click; import gradio; import httpx; print('MinerU + Gradio + httpx imports OK')" && \ ${VENV}/bin/mineru-gradio --help >/dev/null # =========================================================================== diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index f0df181..11e3d18 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -4,7 +4,7 @@ # ============================================================================= services: - # --- WebUI 前端 --- + # --- WebUI 前端（纯前端，不本地处理，通过 Router → Worker 处理）--- gradio: image: mineru-rocm:7.2.1 profiles: ["gradio"] @@ -16,6 +16,7 @@ services: - "10002:7860" environment: - GRADIO_SERVER_NAME=0.0.0.0 + - MINERU_API_BASE=http://mineru-router:8000 - MINERU_MODEL_SOURCE=${MINERU_MODEL_SOURCE:-huggingface} - HF_HUB_CACHE=${HF_HUB_CACHE:-/opt/models/huggingface} - MODELSCOPE_CACHE=${MODELSCOPE_CACHE:-/opt/models/modelscope} @@ -26,10 +27,7 @@ services: - ./scripts:/opt/scripts:ro command: [ - "mineru-gradio", - "--server-name", "0.0.0.0", - "--server-port", "7860", - "--api-url", "http://mineru-router:8000", + "/opt/mineru_venv/bin/python", "/opt/scripts/gradio_client.py", ] depends_on: - router diff --git a/docker/scripts/gradio_client.py b/docker/scripts/gradio_client.py new file mode 100644 index 0000000..c654ec2 --- /dev/null +++ b/docker/scripts/gradio_client.py @@ -0,0 +1,275 @@ +#!/usr/bin/env python3 +"""MinerU Gradio 客户端 — 纯前端，所有解析任务通过 HTTP 转发到 Router/Worker。 + +与 mineru-gradio 的区别： +- 不启动内嵌 FastAPI server +- 不需要 GPU / vLLM +- 通过 HTTP 调用 mineru-router → mineru-worker 处理任务 +""" +import os +import time +import asyncio +import tempfile +import zipfile +import json +from pathlib import Path + +import gradio as gr +import httpx + +API_BASE = os.environ.get("MINERU_API_BASE", "http://mineru-router:8000") +DEFAULT_BACKEND = os.environ.get("MINERU_BACKEND", "hybrid-auto-engine") +DEFAULT_LANG = os.environ.get("MINERU_LANG", "ch") +POLL_INTERVAL = float(os.environ.get("MINERU_POLL_INTERVAL", "1.0")) +MAX_WAIT = float(os.environ.get("MINERU_MAX_WAIT", "600")) + + +async def discover_api(client: httpx.AsyncClient) -> str: + """探测实际可用的 API 前缀: /api/v1/tasks 或 /tasks""" + candidates = [ + f"{API_BASE}/api/v1/health", + f"{API_BASE}/health", + f"{API_BASE}/docs", + ] + for url in candidates: + try: + r = await client.get(url, timeout=5) + if r.status_code < 500: + # FastAPI docs → /api/v1 ; plain health → / + if "/api/v1" in str(r.url): + return "/api/v1/tasks" + return "/tasks" + except Exception: + continue + # 默认新版 API + return "/api/v1/tasks" + + +async def submit_task(client: httpx.AsyncClient, api_prefix: str, + file_path: str, file_name: str, + backend: str, lang: str) -> dict: + """提交解析任务""" + submit_url = f"{API_BASE}{api_prefix}/submit" + with open(file_path, "rb") as f: + files = {"file": (file_name, f, "application/pdf")} + data = { + "backend": backend, + "lang": lang, + } + r = await client.post(submit_url, files=files, data=data, timeout=30) + r.raise_for_status() + return r.json() + + +async def get_task_status(client: httpx.AsyncClient, api_prefix: str, + task_id: str) -> dict: + """查询任务状态""" + url = f"{API_BASE}{api_prefix}/{task_id}" + r = await client.get(url, timeout=10) + r.raise_for_status() + return r.json() + + +async def download_result(client: httpx.AsyncClient, api_prefix: str, + task_id: str, output_dir: str) -> str: + """下载任务结果。尝试多种 endpoint 格式""" + candidates = [ + f"{api_prefix}/{task_id}/data", + f"{api_prefix}/{task_id}/result", + ] + for suffix in candidates: + url = f"{API_BASE}{suffix}" + try: + r = await client.get(url, timeout=60, follow_redirects=True) + if r.status_code == 200: + # 下载 zip 文件 + zip_path = os.path.join(output_dir, f"{task_id}.zip") + with open(zip_path, "wb") as f: + f.write(r.content) + return zip_path + except Exception: + continue + raise RuntimeError(f"Failed to download result for task {task_id}") + + +def extract_readme(zip_path: str, output_dir: str) -> str: + """解压结果并返回 markdown 内容""" + extract_dir = os.path.join(output_dir, "extracted") + os.makedirs(extract_dir, exist_ok=True) + with zipfile.ZipFile(zip_path, "r") as zf: + zf.extractall(extract_dir) + # 查找 markdown 文件 + md_files = list(Path(extract_dir).rglob("*.md")) + if md_files: + return md_files[0].read_text(encoding="utf-8") + return "No markdown output found in result." + + +# --------------------------------------------------------------------------- +# Gradio 处理函数 +# --------------------------------------------------------------------------- +async def process_pdf(file_obj, backend, lang, progress=gr.Progress()): + """Gradio 事件处理：上传 → 提交 → 轮询 → 返回结果""" + if file_obj is None: + return "Please upload a PDF file.", "", "" + + tmp_dir = tempfile.mkdtemp(prefix="mineru-client-") + file_path = file_obj.name if hasattr(file_obj, "name") else os.path.join(tmp_dir, "input.pdf") + file_name = os.path.basename(file_path) + + async with httpx.AsyncClient(timeout=httpx.Timeout(120)) as client: + # 1. 探测 API + progress(0.05, desc="Connecting to API...") + api_prefix = await discover_api(client) + progress(0.1, desc=f"API prefix: {api_prefix}") + + # 2. 提交任务 + progress(0.15, desc="Submitting task...") + try: + submit_resp = await submit_task(client, api_prefix, file_path, file_name, backend, lang) + except httpx.HTTPStatusError as e: + # 尝试旧版 endpoint（不带 /submit 后缀） + alt_data = {"backend": backend, "lang": lang} + try: + alt_url = f"{API_BASE}{api_prefix}" + with open(file_path, "rb") as f: + files = {"file": (file_name, f, "application/pdf")} + r = await client.post(alt_url, files=files, data=alt_data, timeout=30) + r.raise_for_status() + submit_resp = r.json() + except Exception as e2: + return f"Task submission failed: {e}\nAlt attempt: {e2}", "", "" + + task_id = submit_resp.get("task_id") or submit_resp.get("data", {}).get("task_id") + if not task_id: + return f"Unexpected submit response:\n{json.dumps(submit_resp, indent=2)}", "", "" + + progress(0.2, desc=f"Task ID: {task_id}") + + # 3. 轮询状态 + status_text = "" + start_time = time.time() + last_idx = -1 + tickers = ["⣾", "⣽", "⣻", "⢿", "⡿", "⣟", "⣯", "⣷"] + + while True: + elapsed = time.time() - start_time + if elapsed > MAX_WAIT: + return f"Task timed out after {MAX_WAIT}s\nLast status:\n{status_text}", "", "" + + t = (int(elapsed * 2)) % len(tickers) + progress(min(0.2 + 0.6 * (elapsed / 60), 0.8), + desc=f"{tickers[t]} Processing... ({int(elapsed)}s)") + + try: + status_resp = await get_task_status(client, api_prefix, task_id) + except Exception as e: + status_text = f"Polling error: {e}" + await asyncio.sleep(POLL_INTERVAL) + continue + + status = status_resp.get("status", "unknown") + progress_pct = status_resp.get("progress", 0) + err_msg = status_resp.get("error", "") + status_text = json.dumps(status_resp, indent=2, ensure_ascii=False) + + if status == "completed" or status == "success": + progress(0.85, desc="Downloading result...") + break + elif status == "failed": + return f"## Task Failed\n\nError: {err_msg}\n\n```json\n{status_text}\n```", "", "" + elif status == "processing": + pass + else: + pass + + await asyncio.sleep(POLL_INTERVAL) + + # 4. 下载结果 + progress(0.9, desc="Downloading result...") + try: + zip_path = await download_result(client, api_prefix, task_id, tmp_dir) + except Exception as e: + return f"Result download failed: {e}\n\nStatus:\n{status_text}", "", "" + + # 5. 提取 markdown + progress(0.95, desc="Extracting results...") + try: + md_content = extract_readme(zip_path, tmp_dir) + except Exception as e: + md_content = f"Extraction error: {e}" + + progress(1.0, desc="Done!") + return md_content, status_text, zip_path + + +# --------------------------------------------------------------------------- +# Gradio UI +# --------------------------------------------------------------------------- +def create_ui(): + custom_css = """ + .result-panel { min-height: 400px; } + .status-panel { min-height: 200px; font-size: 13px; } + """ + with gr.Blocks(css=custom_css, title="MinerU ROCm — Document Parser") as demo: + gr.Markdown(""" + # MinerU ROCm — Document Parser + + Upload a PDF file. Processing is handled by the **mineru-router → mineru-worker** backend. + """) + + with gr.Row(): + with gr.Column(scale=1): + file_input = gr.File( + label="Upload PDF", + file_types=[".pdf"], + type="filepath", + ) + backend_dd = gr.Dropdown( + label="Backend", + choices=["hybrid-auto-engine", "hybrid-http-client", + "vlm-http-client", "pipeline", "auto"], + value=DEFAULT_BACKEND, + ) + lang_dd = gr.Dropdown( + label="Language", + choices=["ch", "en", "japan", "korean"], + value=DEFAULT_LANG, + ) + submit_btn = gr.Button("Parse Document", variant="primary", size="lg") + status_display = gr.Textbox( + label="Status", + lines=6, + max_lines=12, + elem_classes=["status-panel"], + ) + zip_display = gr.File( + label="Download Result ZIP", + type="filepath", + visible=True, + ) + + with gr.Column(scale=2): + result_display = gr.Markdown( + value="*Upload a PDF to start parsing...*", + elem_classes=["result-panel"], + ) + + submit_btn.click( + fn=process_pdf, + inputs=[file_input, backend_dd, lang_dd], + outputs=[result_display, status_display, zip_display], + ) + + return demo + + +if __name__ == "__main__": + server_name = os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0") + server_port = int(os.environ.get("GRADIO_SERVER_PORT", "7860")) + demo = create_ui() + demo.queue(default_concurrency_limit=3, max_size=10).launch( + server_name=server_name, + server_port=server_port, + share=False, + )