kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: 移除无关文件，并添加 paddlex cli 的脚本

Browse Source
This commit is contained in:
Wenjie Zhang 2025-07-22 00:23:08 +08:00
parent d9a796aec1
commit d50399f217
5 changed files with 86 additions and 447 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
scripts/mineru-api/Dockerfile
View File

@ -1,63 +0,0 @@
# Use the official Ubuntu base image
FROM ubuntu:22.04
# Set environment variables to non-interactive to avoid prompts during installation
# 环境变量设置
ARG http_proxy
ARG https_proxy
ENV http_proxy=$http_proxy \
https_proxy=$https_proxy \
DEBIAN_FRONTEND=noninteractive
# Update the package list and install necessary packages
RUN apt-get update && \
apt-get install -y \
software-properties-common && \
add-apt-repository ppa:deadsnakes/ppa && \
apt-get update && \
apt-get install -y \
python3.10 \
python3.10-venv \
python3.10-distutils \
python3-pip \
wget \
git \
libgl1 \
libreoffice \
fonts-noto-cjk \
fonts-wqy-zenhei \
fonts-wqy-microhei \
ttf-mscorefonts-installer \
fontconfig \
libglib2.0-0 \
libxrender1 \
libsm6 \
libxext6 \
poppler-utils \
&& rm -rf /var/lib/apt/lists/*
# Set Python 3.10 as the default python3
RUN update-alternatives --install /usr/bin/python3 python3 /usr/bin/python3.10 1
# Create a virtual environment for MinerU
RUN python3 -m venv /opt/mineru_venv
# Copy the configuration file template and install magic-pdf latest
RUN /bin/bash -c "source /opt/mineru_venv/bin/activate && \
pip3 install --upgrade pip -i https://mirrors.aliyun.com/pypi/simple && \
pip3 install -U magic-pdf[full] -i https://mirrors.aliyun.com/pypi/simple"
# Download models and update the configuration file
COPY magic-pdf.json /root/magic-pdf.json
RUN /bin/bash -c "pip3 install modelscope -i https://mirrors.aliyun.com/pypi/simple && \
wget https://gcore.jsdelivr.net/gh/opendatalab/MinerU@master/scripts/download_models.py -O download_models.py && \
python3 download_models.py && \
sed -i 's|cpu|cuda|g' /root/magic-pdf.json"
COPY app.py /app/app.py
RUN /bin/bash -c "source /opt/mineru_venv/bin/activate && \
pip3 install fastapi uvicorn python-multipart loguru -i https://mirrors.aliyun.com/pypi/simple"
# Set the entry point to activate the virtual environment and run the command line tool
ENTRYPOINT ["/bin/bash", "-c", "source /opt/mineru_venv/bin/activate && exec \"$@\"", "--"]

307
scripts/mineru-api/app.py
View File

@ -1,307 +0,0 @@
import json
import os
import tempfile
from base64 import b64encode
from glob import glob
from io import StringIO
import magic_pdf.model as model_config
import uvicorn
from fastapi import FastAPI, Form, UploadFile
from fastapi.responses import JSONResponse
from loguru import logger
from magic_pdf.config.enums import SupportedPdfParseMethod
from magic_pdf.data.data_reader_writer import DataWriter, FileBasedDataWriter
from magic_pdf.data.data_reader_writer.s3 import S3DataReader, S3DataWriter
from magic_pdf.data.dataset import ImageDataset, PymuDocDataset
from magic_pdf.data.read_api import read_local_images, read_local_office
from magic_pdf.libs.config_reader import get_bucket_name, get_s3_config
from magic_pdf.model.doc_analyze_by_custom_model import doc_analyze
from magic_pdf.operators.models import InferenceResult
from magic_pdf.operators.pipes import PipeResult
model_config.__use_inside_model__ = True
app = FastAPI()
pdf_extensions = [".pdf"]
office_extensions = [".ppt", ".pptx", ".doc", ".docx"]
image_extensions = [".png", ".jpg", ".jpeg"]
class MemoryDataWriter(DataWriter):
def __init__(self):
self.buffer = StringIO()
def write(self, path: str, data: bytes) -> None:
if isinstance(data, str):
self.buffer.write(data)
else:
self.buffer.write(data.decode("utf-8"))
def write_string(self, path: str, data: str) -> None:
self.buffer.write(data)
def get_value(self) -> str:
return self.buffer.getvalue()
def close(self):
self.buffer.close()
def init_writers(
file_path: str = None,
file: UploadFile = None,
output_path: str = None,
output_image_path: str = None,
) -> tuple[
S3DataWriter | FileBasedDataWriter,
S3DataWriter | FileBasedDataWriter,
bytes,
]:
"""
Initialize writers based on path type
Args:
file_path: file path (local path or S3 path)
file: Uploaded file object
output_path: Output directory path
output_image_path: Image output directory path
Returns:
Tuple[writer, image_writer, file_bytes]: Returns initialized writer tuple and file content
"""
file_extension:str = None
if file_path:
is_s3_path = file_path.startswith("s3://")
if is_s3_path:
bucket = get_bucket_name(file_path)
ak, sk, endpoint = get_s3_config(bucket)
writer = S3DataWriter(
output_path, bucket=bucket, ak=ak, sk=sk, endpoint_url=endpoint
)
image_writer = S3DataWriter(
output_image_path, bucket=bucket, ak=ak, sk=sk, endpoint_url=endpoint
)
# 临时创建reader读取文件内容
temp_reader = S3DataReader(
"", bucket=bucket, ak=ak, sk=sk, endpoint_url=endpoint
)
file_bytes = temp_reader.read(file_path)
file_extension = os.path.splitext(file_path)[1]
else:
writer = FileBasedDataWriter(output_path)
image_writer = FileBasedDataWriter(output_image_path)
os.makedirs(output_image_path, exist_ok=True)
with open(file_path, "rb") as f:
file_bytes = f.read()
file_extension = os.path.splitext(file_path)[1]
else:
# 处理上传的文件
file_bytes = file.file.read()
file_extension = os.path.splitext(file.filename)[1]
writer = FileBasedDataWriter(output_path)
image_writer = FileBasedDataWriter(output_image_path)
os.makedirs(output_image_path, exist_ok=True)
return writer, image_writer, file_bytes, file_extension
def process_file(
file_bytes: bytes,
file_extension: str,
parse_method: str,
image_writer: S3DataWriter | FileBasedDataWriter,
) -> tuple[InferenceResult, PipeResult]:
"""
Process PDF file content
Args:
file_bytes: Binary content of file
file_extension: file extension
parse_method: Parse method ('ocr', 'txt', 'auto')
image_writer: Image writer
Returns:
Tuple[InferenceResult, PipeResult]: Returns inference result and pipeline result
"""
ds: PymuDocDataset | ImageDataset = None
if file_extension in pdf_extensions:
ds = PymuDocDataset(file_bytes)
elif file_extension in office_extensions:
# 需要使用office解析
temp_dir = tempfile.mkdtemp()
with open(os.path.join(temp_dir, f"temp_file.{file_extension}"), "wb") as f:
f.write(file_bytes)
ds = read_local_office(temp_dir)[0]
elif file_extension in image_extensions:
# 需要使用ocr解析
temp_dir = tempfile.mkdtemp()
with open(os.path.join(temp_dir, f"temp_file.{file_extension}"), "wb") as f:
f.write(file_bytes)
ds = read_local_images(temp_dir)[0]
infer_result: InferenceResult = None
pipe_result: PipeResult = None
if parse_method == "ocr":
infer_result = ds.apply(doc_analyze, ocr=True)
pipe_result = infer_result.pipe_ocr_mode(image_writer)
elif parse_method == "txt":
infer_result = ds.apply(doc_analyze, ocr=False)
pipe_result = infer_result.pipe_txt_mode(image_writer)
else: # auto
if ds.classify() == SupportedPdfParseMethod.OCR:
infer_result = ds.apply(doc_analyze, ocr=True)
pipe_result = infer_result.pipe_ocr_mode(image_writer)
else:
infer_result = ds.apply(doc_analyze, ocr=False)
pipe_result = infer_result.pipe_txt_mode(image_writer)
return infer_result, pipe_result
def encode_image(image_path: str) -> str:
"""Encode image using base64"""
with open(image_path, "rb") as f:
return b64encode(f.read()).decode()
@app.get("/health", tags=["health"], summary="Health check endpoint")
async def health_check():
"""
Simple health check endpoint to confirm the API is running.
"""
return JSONResponse(content={"status": "healthy"}, status_code=200)
@app.post(
"/file_parse",
tags=["projects"],
summary="Parse files (supports local files and S3)",
)
async def file_parse(
file: UploadFile = None,
file_path: str = Form(None),
parse_method: str = Form("auto"),
is_json_md_dump: bool = Form(False),
output_dir: str = Form("output"),
return_layout: bool = Form(False),
return_info: bool = Form(False),
return_content_list: bool = Form(False),
return_images: bool = Form(False),
):
"""
Execute the process of converting PDF to JSON and MD, outputting MD and JSON files
to the specified directory.
Args:
file: The PDF file to be parsed. Must not be specified together with
`file_path`
file_path: The path to the PDF file to be parsed. Must not be specified together
with `file`
parse_method: Parsing method, can be auto, ocr, or txt. Default is auto. If
results are not satisfactory, try ocr
is_json_md_dump: Whether to write parsed data to .json and .md files. Default
to False. Different stages of data will be written to different .json files
(3 in total), md content will be saved to .md file
output_dir: Output directory for results. A folder named after the PDF file
will be created to store all results
return_layout: Whether to return parsed PDF layout. Default to False
return_info: Whether to return parsed PDF info. Default to False
return_content_list: Whether to return parsed PDF content list. Default to False
"""
try:
if (bool(file) == bool(file_path)):
return JSONResponse(
content={"error": "Must provide either file or file_path"},
status_code=400,
)
# Get PDF filename
file_name = os.path.basename(file_path if file_path else file.filename).split(
"."
)[0]
output_path = f"{output_dir}/{file_name}"
output_image_path = f"{output_path}/images"
# Initialize readers/writers and get PDF content
writer, image_writer, file_bytes, file_extension = init_writers(
file_path=file_path,
file=file,
output_path=output_path,
output_image_path=output_image_path,
)
# Process PDF
infer_result, pipe_result = process_file(file_bytes, file_extension, parse_method, image_writer)
# Use MemoryDataWriter to get results
content_list_writer = MemoryDataWriter()
md_content_writer = MemoryDataWriter()
middle_json_writer = MemoryDataWriter()
# Use PipeResult's dump method to get data
pipe_result.dump_content_list(content_list_writer, "", "images")
pipe_result.dump_md(md_content_writer, "", "images")
pipe_result.dump_middle_json(middle_json_writer, "")
# Get content
content_list = json.loads(content_list_writer.get_value())
md_content = md_content_writer.get_value()
middle_json = json.loads(middle_json_writer.get_value())
model_json = infer_result.get_infer_res()
# If results need to be saved
if is_json_md_dump:
writer.write_string(
f"{file_name}_content_list.json", content_list_writer.get_value()
)
writer.write_string(f"{file_name}.md", md_content)
writer.write_string(
f"{file_name}_middle.json", middle_json_writer.get_value()
)
writer.write_string(
f"{file_name}_model.json",
json.dumps(model_json, indent=4, ensure_ascii=False),
)
# Save visualization results
pipe_result.draw_layout(os.path.join(output_path, f"{file_name}_layout.pdf"))
pipe_result.draw_span(os.path.join(output_path, f"{file_name}_spans.pdf"))
pipe_result.draw_line_sort(
os.path.join(output_path, f"{file_name}_line_sort.pdf")
)
infer_result.draw_model(os.path.join(output_path, f"{file_name}_model.pdf"))
# Build return data
data = {}
if return_layout:
data["layout"] = model_json
if return_info:
data["info"] = middle_json
if return_content_list:
data["content_list"] = content_list
if return_images:
image_paths = glob(f"{output_image_path}/*.jpg")
data["images"] = {
os.path.basename(
image_path
): f"data:image/jpeg;base64,{encode_image(image_path)}"
for image_path in image_paths
}
data["md_content"] = md_content # md_content is always returned
# Clean up memory writers
content_list_writer.close()
md_content_writer.close()
middle_json_writer.close()
return JSONResponse(data, status_code=200)
except Exception as e:
logger.exception(e)
return JSONResponse(content={"error": str(e)}, status_code=500)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8888)

44
scripts/mineru-api/magic-pdf.json
View File

@ -1,44 +0,0 @@
{
"bucket_info":{
"bucket-name-1":["ak", "sk", "endpoint"],
"bucket-name-2":["ak", "sk", "endpoint"]
},
"models-dir":"/opt/models",
"layoutreader-model-dir":"/opt/layoutreader",
"device-mode":"cuda",
"layout-config": {
"model": "doclayout_yolo"
},
"formula-config": {
"mfd_model": "yolo_v8_mfd",
"mfr_model": "unimernet_small",
"enable": true
},
"table-config": {
"model": "rapid_table",
"sub_model": "slanet_plus",
"enable": true,
"max_time": 400
},
"llm-aided-config": {
"formula_aided": {
"api_key": "your_api_key",
"base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
"model": "qwen2.5-7b-instruct",
"enable": false
},
"text_aided": {
"api_key": "your_api_key",
"base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
"model": "qwen2.5-7b-instruct",
"enable": false
},
"title_aided": {
"api_key": "your_api_key",
"base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
"model": "qwen2.5-32b-instruct",
"enable": false
}
},
"config_version": "1.2.0"
}

32
scripts/paddlestructure/example_usage.py
View File

@ -1,32 +0,0 @@
#!/usr/bin/env python3
"""
PaddleX 文档分析使用示例
演示如何使用 analyze_document 函数分析文档
"""
from src.plugins.paddlex import analyze_document
import json
def analyze_custom_file(file_path: str):
"""分析自定义文件的示例函数"""
print(f"\n🔍 分析自定义文件: {file_path}")
result = analyze_document(file_path=file_path, base_url="http://172.19.13.5:8080")
if result["success"]:
return result
else:
print(f"❌ 分析失败: {result['error']}")
return None
if __name__ == "__main__":
# main()
# 如果您想分析其他文件，可以取消注释下面的代码
custom_file = "test/data/ocr_test/1706.03762v7_扫描版.pdf"
custom_file = "test/data/ocr_test/PixPin_2025-06-19_23-42-17.png"
print(analyze_custom_file(custom_file))

87
src/plugins/paddlex.py
View File

@ -5,7 +5,13 @@ import base64
import os
import time
from typing import Optional, Any
from src.utils import logger
from pathlib import Path
if __name__ == "__main__":
from loguru import logger
import typer
else:
from src.utils import logger
@ -297,3 +303,82 @@ def analyze_document(file_path: str, base_url: str = "http://localhost:8080") ->
def check_paddlex_health(base_url: str = "http://localhost:8080") -> bool:
return requests.get(f"{base_url}/health", timeout=5)
def analyze_folder(input_dir: str, output_dir: str, base_url: str = "http://localhost:8080"):
"""分析文件夹中的所有支持文件，保存为txt格式"""
input_path = Path(input_dir)
output_path = Path(output_dir)
if not input_path.exists():
print(f"❌ 输入目录不存在：{input_dir}")
return
# 创建输出目录
output_path.mkdir(parents=True, exist_ok=True)
# 获取所有支持的文件
supported_extensions = {'.pdf', '.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.tif'}
files = []
for root, dirs, filenames in os.walk(input_dir):
for filename in filenames:
file_path = Path(root) / filename
if file_path.suffix.lower() in supported_extensions:
files.append(file_path)
if not files:
print(f"⚠️ 没有找到支持的文件")
return
print(f"📁 找到 {len(files)} 个文件")
success_count = 0
for i, file_path in enumerate(files, 1):
print(f"🔄 [{i}/{len(files)}] {file_path.name}")
try:
# 分析文档
result = analyze_document(str(file_path), base_url)
if result.get("success"):
# 保持目录结构
relative_path = file_path.relative_to(input_path)
output_file = output_path / relative_path.with_suffix('.txt')
output_file.parent.mkdir(parents=True, exist_ok=True)
# 写入文本内容
text_content = result.get("full_text", "未提取到内容") if result.get("success") else f"分析失败：{result.get('error', '未知错误')}"
with open(output_file, 'w', encoding='utf-8') as f:
f.write(text_content)
success_count += 1
print(f"✅ {output_file.name}")
else:
print(f"❌ 失败: {result.get('error')}")
except Exception as e:
print(f"❌ 异常: {str(e)}")
time.sleep(0.5)
print(f"\n📊 完成！成功: {success_count}, 总计: {len(files)}")
if __name__ == "__main__":
app = typer.Typer(help="PaddleX 文档分析工具")
@app.command()
def single(file_path: str, base_url: str = "http://172.19.13.5:8080"):
"""分析单个文件"""
result = analyze_document(file_path, base_url)
if result["success"]:
print(f"✅ 成功提取 {len(result['full_text'])} 个字符")
else:
print(f"❌ 失败: {result.get('error')}")
@app.command()
def folder(input_dir: str, output_dir: str, base_url: str = "http://172.19.13.5:8080"):
"""批量分析文件夹"""
analyze_folder(input_dir, output_dir, base_url)
app()