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

feat: 添加 PaddleX 服务支持，更新 Docker 配置并增加文档分析示例

Browse Source
This commit is contained in:
Wenjie Zhang 2025-06-20 17:33:14 +08:00
parent 6671c920c3
commit 5657a250bb
4 changed files with 368 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
docker-compose.yml
View File

@ -33,6 +33,7 @@ services:
- NEO4J_PASSWORD=${NEO4J_PASSWORD:-0123456789}
- MILVUS_URI=http://milvus:19530
- MINERU_OCR_URI=http://mineru:30000
- PADDLEX_URI=http://paddlex:8080
- MODEL_DIR=/models
- RUNNING_IN_DOCKER=true
command: uv run uvicorn server.main:app --host 0.0.0.0 --port 5050 --reload
@ -183,7 +184,36 @@ services:
reservations:
devices:
- driver: nvidia
device_ids: ["0"]
device_ids: ["1"]
capabilities: [gpu]
networks:
- app-network
restart: unless-stopped
paddlex:
build:
context: .
dockerfile: docker/paddlex.Dockerfile
image: paddlex:latest
container_name: paddlex
profiles:
- all
volumes:
- ./docker/volumes/paddlex:/paddlex
ports:
- "8080:8080"
healthcheck:
test: ["CMD-SHELL", "curl -f http://localhost:8080/ || exit 1"]
interval: 20s
timeout: 10s
retries: 5
start_period: 60s
deploy:
resources:
reservations:
devices:
- driver: nvidia
device_ids: ['0']
capabilities: [gpu]
networks:
- app-network

12
docker/paddlex.Dockerfile Normal file
View File

@ -0,0 +1,12 @@
FROM ccr-2vdh3abv-pub.cnc.bj.baidubce.com/paddlex/paddlex:paddlex3.0.1-paddlepaddle3.0.0-gpu-cuda11.8-cudnn8.9-trt8.6
WORKDIR /paddle
# 安装 hpi-cpu，如您所指示
RUN paddlex --install hpi-cpu
# 暴露 PaddleX 服务端口
EXPOSE 8080
# 运行 PaddleX PP-StructureV3 流水线服务
CMD ["paddlex", "--serve", "--pipeline", "PP-StructureV3", "--host", "0.0.0.0", "--port", "8080"]

32
scripts/paddlestructure/example_usage.py Normal file
View File

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

293
scripts/paddlestructure/paddlex_layout_parser.py Normal file
View File

@ -0,0 +1,293 @@
import requests
import json
import base64
import os
import time
from typing import Optional, Dict, Any
class PaddleXLayoutParser:
"""PaddleX 版面解析服务客户端"""
def __init__(self, base_url: str = "http://localhost:8080"):
self.base_url = base_url.rstrip('/')
self.endpoint = f"{self.base_url}/layout-parsing"
def encode_file_to_base64(self, file_path: str) -> str:
with open(file_path, 'rb') as file:
encoded = base64.b64encode(file.read()).decode('utf-8')
return encoded
def _process_file_input(self, file_input: str) -> str:
# 检查是否为本地文件路径
if os.path.exists(file_input):
print(f"📁 检测到本地文件: {file_input}")
print(f"📏 文件大小: {os.path.getsize(file_input) / 1024 / 1024:.2f} MB")
try:
# 将本地文件编码为Base64
encoded_content = self.encode_file_to_base64(file_input)
print(f"✅ 文件已编码为Base64，长度: {len(encoded_content)} 字符")
return encoded_content
except Exception as e:
print(f"❌ 文件编码失败: {e}")
raise
# 检查是否为URL
elif file_input.startswith(('http://', 'https://')):
print(f"🌐 检测到URL: {file_input}")
return file_input
# 否则假设为Base64编码内容
else:
print(f"📝 假设为Base64编码内容，长度: {len(file_input)} 字符")
return file_input
def layout_parsing(self,
file_input: str,
file_type: Optional[int] = None,
use_textline_orientation: Optional[bool] = None,
use_seal_recognition: Optional[bool] = None,
use_table_recognition: Optional[bool] = None,
use_formula_recognition: Optional[bool] = None,
use_chart_recognition: Optional[bool] = None,
use_region_detection: Optional[bool] = None,
layout_threshold: Optional[float] = None,
layout_nms: Optional[bool] = None,
use_doc_orientation_classify: Optional[bool] = True,
use_doc_unwarping: Optional[bool] = False,
use_wired_table_cells_trans_to_html: Optional[bool] = True, # 是否启用无有线表单元格检测结果直转HTML，默认False，启用则直接基于有线表单元格检测结果的几何关系构建HTML。
**kwargs) -> Dict[str, Any]:
"""
调用版面解析API：https://paddlepaddle.github.io/PaddleX/latest/pipeline_usage/tutorials/ocr_pipelines/PP-StructureV3.html#22-python
"""
# 处理文件输入：检测是否为本地文件路径
processed_file_input = self._process_file_input(file_input)
payload = {"file": processed_file_input}
# 添加可选参数
optional_params = {
"fileType": file_type,
"useDocOrientationClassify": use_doc_orientation_classify,
"useDocUnwarping": use_doc_unwarping,
"useTextlineOrientation": use_textline_orientation,
"useSealRecognition": use_seal_recognition,
"useTableRecognition": use_table_recognition,
"useFormulaRecognition": use_formula_recognition,
"useChartRecognition": use_chart_recognition,
"useRegionDetection": use_region_detection,
"layoutThreshold": layout_threshold,
"layoutNms": layout_nms,
"useWiredTableCellsTransToHtml": use_wired_table_cells_trans_to_html,
}
# 添加非空参数
for key, value in optional_params.items():
if value is not None:
payload[key] = value
# 添加其他kwargs参数
for key, value in kwargs.items():
if value is not None:
payload[key] = value
try:
response = requests.post(
self.endpoint,
json=payload,
headers={"Content-Type": "application/json"},
timeout=300
)
if response.status_code == 200:
result = response.json()
print("✅ 请求成功!")
return result
else:
print("❌ 请求失败!")
try:
error_result = response.json()
print(f"错误信息: {json.dumps(error_result, indent=2, ensure_ascii=False)}")
return error_result
except:
print(f"响应内容: {response.text}")
return {"error": response.text, "status_code": response.status_code}
except requests.exceptions.RequestException as e:
print(f"❌ 网络请求异常: {e}")
return {"error": str(e)}
except Exception as e:
print(f"❌ 其他异常: {e}")
return {"error": str(e)}
def _parse_recognition_result(api_result: Dict[str, Any], file_path: str) -> Dict[str, Any]:
# 基本信息
parsed_result = {
"success": True,
"file_path": file_path,
"file_name": os.path.basename(file_path),
"log_id": api_result.get("logId"),
"total_pages": 0,
"pages": [],
"full_text": "",
"summary": {}
}
result_data = api_result.get("result", {})
layout_results = result_data.get("layoutParsingResults", [])
data_info = result_data.get("dataInfo", {})
# 数据信息
parsed_result["total_pages"] = len(layout_results)
parsed_result["document_info"] = {
"type": data_info.get("type", "unknown"),
"total_pages": data_info.get("numPages", len(layout_results)),
"page_dimensions": data_info.get("pages", [])
}
# 统计信息
total_elements = 0
total_tables = 0
total_formulas = 0
total_charts = 0
total_seals = 0
all_text_content = []
# 解析每页结果
for page_index, page_result in enumerate(layout_results):
page_info = {
"page_number": page_index + 1,
"content": {},
"statistics": {}
}
# Markdown内容
if "markdown" in page_result:
markdown = page_result["markdown"]
page_info["content"]["markdown_text"] = markdown.get("text", "")
page_info["content"]["images"] = list(markdown.get("images", {}).keys())
page_info["content"]["is_paragraph_start"] = markdown.get("isStart", False)
page_info["content"]["is_paragraph_end"] = markdown.get("isEnd", False)
# 收集文本内容
if markdown.get("text"):
all_text_content.append(markdown["text"])
# 详细识别结果
if "prunedResult" in page_result:
pruned = page_result["prunedResult"]
# 版面检测
layout_detection = pruned.get("layout_detection", [])
page_info["statistics"]["layout_elements"] = len(layout_detection)
total_elements += len(layout_detection)
# OCR结果
ocr_result = pruned.get("ocr_result", [])
page_info["statistics"]["ocr_elements"] = len(ocr_result)
# 表格识别
table_result = pruned.get("table_result", [])
page_info["statistics"]["tables"] = len(table_result)
total_tables += len(table_result)
# 公式识别
formula_result = pruned.get("formula_result", [])
page_info["statistics"]["formulas"] = len(formula_result)
total_formulas += len(formula_result)
# 图表识别
chart_result = pruned.get("chart_result", [])
page_info["statistics"]["charts"] = len(chart_result)
total_charts += len(chart_result)
# 印章识别
seal_result = pruned.get("seal_result", [])
page_info["statistics"]["seals"] = len(seal_result)
total_seals += len(seal_result)
# 详细元素信息
page_info["content"]["layout_elements"] = layout_detection
page_info["content"]["ocr_elements"] = ocr_result
page_info["content"]["tables"] = table_result
page_info["content"]["formulas"] = formula_result
page_info["content"]["charts"] = chart_result
page_info["content"]["seals"] = seal_result
parsed_result["pages"].append(page_info)
# 汇总全文内容
parsed_result["full_text"] = "\n\n".join(all_text_content)
# 汇总统计信息
parsed_result["summary"] = {
"total_elements": total_elements,
"total_tables": total_tables,
"total_formulas": total_formulas,
"total_charts": total_charts,
"total_seals": total_seals,
"total_characters": len(parsed_result["full_text"]),
"average_elements_per_page": round(total_elements / max(1, len(layout_results)), 2)
}
return parsed_result
def analyze_document(file_path: str) -> Dict[str, Any]:
# 检查文件是否存在
if not os.path.exists(file_path):
return {
"success": False,
"error": f"文件不存在: {file_path}",
"file_path": file_path
}
# 初始化客户端
client = PaddleXLayoutParser()
# 判断文件类型
file_ext = os.path.splitext(file_path)[1].lower()
if file_ext == '.pdf':
file_type = 0
elif file_ext in ['.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.tif']:
file_type = 1
else:
return {
"success": False,
"error": f"不支持的文件类型: {file_ext}",
"file_path": file_path
}
print(f"📄 开始分析文档: {os.path.basename(file_path)}")
print(f"📏 文件大小: {os.path.getsize(file_path) / 1024 / 1024:.2f} MB")
print(f"📋 文件类型: {'PDF' if file_type == 0 else '图片'}")
try:
# 调用API进行识别
result = client.layout_parsing(file_input=file_path, file_type=file_type)
# 检查API调用是否成功
if result.get("errorCode") != 0:
return {
"success": False,
"error": result.get("errorMsg", "API调用失败"),
"file_path": file_path,
"raw_result": result
}
# 解析结果
analysis_result = _parse_recognition_result(result, file_path)
return analysis_result
except Exception as e:
return {
"success": False,
"error": f"处理异常: {str(e)}",
"file_path": file_path
}