LQAgentPlatform/utils_test/MinerU_Test/mineru_test.py

# -*- coding: utf-8 -*-
import sys
import os
import json
import base64
import urllib.request
# 获取当前脚本的绝对路径
current_file = os.path.abspath(__file__)
# 获取脚本所在目录（MinerU_Test）
current_dir = os.path.dirname(current_file)
# 获取上一级目录（utils_test）
parent_dir = os.path.dirname(current_dir)
root_dir = os.path.dirname(parent_dir)



import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from foundation.observability.monitoring.time_statistics import track_execution_time

@track_execution_time
def save_images_from_response(response_data, output_dir="test/mineru_temp"):
    """从API响应中提取并保存图片"""
    # 创建输出目录
    os.makedirs(output_dir, exist_ok=True)

    # 检查是否有图片数据
    if "results" not in response_data or "scan" not in response_data["results"]:
        print("❌ 响应中未找到图片数据")
        return

    scan_result = response_data["results"]["scan"]
    saved_count = 0

    # 保存base64图片
    if "images" in scan_result:
        print(f"发现 {len(scan_result['images'])} 个base64图片")
        for filename, base64_data in scan_result["images"].items():
            try:
                # 解析base64数据格式
                if base64_data.startswith("data:image/"):
                    # 提取实际的base64数据
                    header, base64_string = base64_data.split(",", 1)
                    file_extension = header.split("/")[1].split(";")[0]

                    # 解码base64
                    image_data = base64.b64decode(base64_string)

                    # 保存图片文件
                    output_filename = f"table_{saved_count}.{file_extension}"
                    output_path = os.path.join(output_dir, output_filename)

                    with open(output_path, "wb") as f:
                        f.write(image_data)

                    print(f"✅ 保存图片: {output_filename} ({len(image_data)} bytes)")
                    saved_count += 1

                else:
                    # 直接base64数据
                    image_data = base64.b64decode(base64_data)
                    output_path = os.path.join(output_dir, f"table_{saved_count}.jpg")

                    with open(output_path, "wb") as f:
                        f.write(image_data)

                    print(f"✅ 保存图片: table_{saved_count}.jpg ({len(image_data)} bytes)")
                    saved_count += 1

            except Exception as e:
                print(f"❌ 保存图片 {filename} 失败: {e}")

    # 检查middle_json中的图片信息（如果有额外的）
    if "middle_json" in scan_result:
        try:
            middle_json = json.loads(scan_result["middle_json"])

            # 查找图片路径和HTML信息
            for pdf_info in middle_json.get("pdf_info", []):
                for preproc_block in pdf_info.get("preproc_blocks", []):
                    if preproc_block.get("type") == "table":
                        for block in preproc_block.get("blocks", []):
                            if block.get("type") == "table_body":
                                for line in block.get("lines", []):
                                    for span in line.get("spans", []):
                                        if span.get("type") == "table":
                                            # 保存HTML内容
                                            html_content = span.get('html', 'N/A')
                                            if html_content != 'N/A':
                                                html_file = os.path.join(output_dir, f"table_html_{saved_count}.html")
                                                with open(html_file, "w", encoding="utf-8") as f:
                                                    f.write(html_content)
                                                print(f"✅ 保存表格HTML: table_html_{saved_count}.html")

        except json.JSONDecodeError as e:
            print(f"❌ 解析middle_json失败: {e}")

    print(f"✅ 图片保存完成，共保存 {saved_count} 个图片文件到 {output_dir}")
    return saved_count

@track_execution_time
def save_middle_json(response_data, output_dir="test/mineru_temp"):
    """提取并保存middle_json为单独文件"""
    os.makedirs(output_dir, exist_ok=True)

    if "results" not in response_data or "scan" not in response_data["results"]:
        print("❌ 响应中未找到middle_json数据")
        return False

    scan_result = response_data["results"]["scan"]
    if "middle_json" not in scan_result:
        print("❌ scan结果中未找到middle_json")
        return False

    try:
        # 解析middle_json字符串
        middle_json_data = json.loads(scan_result["middle_json"])

        # 保存格式化的middle_json
        middle_json_file = os.path.join(output_dir, "middle_json_pretty.json")
        with open(middle_json_file, "w", encoding="utf-8") as f:
            json.dump(middle_json_data, f, ensure_ascii=False, indent=4)

        print(f"✅ Middle JSON已保存到 {middle_json_file}")

        # 打印middle_json的基本信息
        if "pdf_info" in middle_json_data:
            pdf_info = middle_json_data["pdf_info"]
            print(f"📄 PDF信息: {len(pdf_info)} 页")

            # 统计各类型块的数量
            total_blocks = 0
            table_count = 0
            text_count = 0
            title_count = 0

            for page_info in pdf_info:
                preproc_blocks = page_info.get("preproc_blocks", [])
                total_blocks += len(preproc_blocks)

                for block in preproc_blocks:
                    block_type = block.get("type", "")
                    if block_type == "table":
                        table_count += 1
                    elif block_type == "text":
                        text_count += 1
                    elif block_type == "title":
                        title_count += 1

            print(f"📊 统计信息:")
            print(f"   总块数: {total_blocks}")
            print(f"   表格块: {table_count}")
            print(f"   文本块: {text_count}")
            print(f"   标题块: {title_count}")

        return True

    except json.JSONDecodeError as e:
        print(f"❌ 解析middle_json失败: {e}")
        return False
    except Exception as e:
        print(f"❌ 保存middle_json失败: {e}")
        return False

@track_execution_time
def parse_file():
    """使用 API 解析文件"""
    API_URL = "http://aiclu.small-app.wang:8000/file_parse"

    files = [
        ("files", ("scan.png", open("utils_test\MinerU_Test\sgfa_mineru_testimage.png", "rb"), "image/png")),
    ]

    data = {
        "backend": "pipeline",
        "parse_method": "auto",
        "formula_enable": "true",
        "table_enable": "true",
        "return_md": "true",
        "start_page_id": 0,
        "end_page_id": None,
        "return_middle_json": "true",
        "return_images": "true",
        "response_format_zip": "true",  # 保持为 true
    }

    response = requests.post(API_URL, files=files, data=data)

    # 检查响应是否为 ZIP 包（通过响应头或内容判断）
    if "application/zip" in response.headers.get("Content-Type", "") or response.content.startswith(b"PK"):
        # 保存 ZIP 包到本地
        zip_output_dir = "utils_test\MinerU_Test\mineru_temp"
        os.makedirs(zip_output_dir, exist_ok=True)
        zip_file_path = os.path.join(zip_output_dir, "api_response.zip")

        # 保存二进制 ZIP 数据
        with open(zip_file_path, "wb") as f:
            f.write(response.content)
        print(f"✅ ZIP 包已保存到: {zip_file_path} ({len(response.content)} bytes)")

        # 自动解压 ZIP 包
        import zipfile
        with zipfile.ZipFile(zip_file_path, "r") as zip_ref:
            zip_ref.extractall(os.path.join(zip_output_dir, "unzipped"))
        print(f"✅ ZIP 包已解压到: {os.path.join(zip_output_dir, 'unzipped')}")

        # 解压后可以读取里面的 JSON/图片文件（示例：读取 middle_json）
        unzipped_dir = os.path.join(zip_output_dir, "unzipped")
        middle_json_path = os.path.join(unzipped_dir, "middle_json.json")  # 假设 ZIP 内有这个文件（需根据实际 ZIP 结构调整）
        if os.path.exists(middle_json_path):
            with open(middle_json_path, "r", encoding="utf-8") as f:
                middle_json_data = json.load(f)
            print(f"📊 解压后读取到 middle_json，包含 {len(middle_json_data.get('pdf_info', []))} 页 PDF 信息")

    else:
        # 若 API 未返回 ZIP（兼容情况），尝试按 JSON 解析
        try:
            result = response.json()
            print("=== API响应概要（JSON格式）===")
            # 后续逻辑和原代码一致...
        except json.JSONDecodeError as e:
            print(f"❌ 解析失败：既不是 ZIP 也不是 JSON - {e}")

parse_file()