FORMAT_CONVERSION_MAXCOMPUTE/format_convert/convert_test.py

import base64
import json
import os
import random
import sys
import time
from glob import glob
sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/../")
from format_convert.utils import get_platform, request_post, get_md5_from_bytes
from format_convert.convert import to_html
import multiprocessing as mp


def test_one(p, page_no_range=None, from_remote=False):
    start_time = time.time()
    with open(p, "rb") as f:
        file_bytes = f.read()
    file_base64 = base64.b64encode(file_bytes)

    _md5 = get_md5_from_bytes(file_bytes)

    data = {"file": file_base64, "type": p.split(".")[-1], "filemd5": _md5, 'page_no': page_no_range}
    if from_remote:
        _url = 'http://121.46.18.113:15010/convert'
        # _url = 'http://192.168.2.103:15010/convert'
        # _url = 'http://192.168.2.102:15011/convert'
        # _url = 'http://172.16.160.65:15010/convert'
        # _url = 'http://127.0.0.1:15010/convert'
        result = json.loads(request_post(_url, data, time_out=10000))
        text_str = ""
        for t in result.get("result_html"):
            text_str += t
        to_html(os.path.dirname(os.path.abspath(__file__)) + "/../result.html",
                text_str)
    else:
        print("only support remote!")

    print(_md5)
    print('第', page_no_range.split(',')[0], '页到第', page_no_range.split(',')[-1], '页')
    print("result_text", result.get("result_text")[0][:20])
    print("is_success", result.get("is_success"))
    print(time.time()-start_time)


def test_duplicate(path_list, process_no=None):
    start_time = time.time()
    # random.shuffle(path_list)
    for i in range(10):
        if i % 10 == 0:
            if process_no is not None:
                print("Process", process_no, i*len(path_list), time.time()-start_time)
            else:
                print("Loop", i*len(path_list), time.time()-start_time)
        for p in path_list:
            test_one(p, from_remote=True)


def test_maxcompute(p, page_no_range=None):
    from format_convert import convert
    start_time = time.time()
    with open(p, "rb") as f:
        file_bytes = f.read()
    file_base64 = base64.b64encode(file_bytes)
    _md5 = get_md5_from_bytes(file_bytes)

    data = {"file": file_base64, "type": p.split(".")[-1], "filemd5": _md5, 'page_no': page_no_range}
    result = convert.convert(data)
    text_str = ""
    for t in result.get("result_html"):
        text_str += t
    to_html(os.path.dirname(os.path.abspath(__file__)) + "/../result.html",
            text_str)

    print(_md5)
    print('第', page_no_range.split(',')[0], '页到第', page_no_range.split(',')[-1], '页')
    print("result_text", result.get("result_text")[0][:20])
    print("is_success", result.get("is_success"))
    print(time.time()-start_time)


if __name__ == '__main__':
    if get_platform() == "Windows":
        # file_path = "C:/Users/Administrator/Desktop/2.png"
        file_path = "C:/Users/Administrator/Desktop/test_xls/error4.xls"
        # file_path = "C:/Users/Administrator/Desktop/test_doc/error5.doc"
        # file_path = "D:/BIDI_DOC/比地_文档/1677829036789.pdf"
        # file_path = "D:/BIDI_DOC/比地_文档/2022/Test_ODPS/1624325845476.pdf"
        # file_path = "C:/Users/Administrator/Downloads/1688432101601.xlsx"
        # file_path = "C:/Users/Administrator/Desktop/test_doc/error14.docx"
        # file_path = "C:/Users/Administrator/Desktop/test_image/error36.png"
        # file_path = "C:/Users/Administrator/Desktop/test_b_table/error1.png"
        # file_path = "C:/Users/Administrator/Desktop/test_pdf/表格连接error/error7.pdf"
        # file_path = "C:/save_b_table/0-0895e32470613dd7be1139eefd1342c4.png"
    else:
        file_path = "1660296734009.pdf"

    test_one(file_path, page_no_range='1,-1', from_remote=True)

    file_path = "C:/Users/Administrator/Downloads/"
    # file_path = r"C:\Users\Administrator\Desktop\test_pdf\直接读表格线error/"
    # file_path = r"C:\Users\Administrator\Desktop\test_pdf\表格连接error/"
    test_pdf_list = [['6df7f2bd5e8cac99a15a6c012e0d82a8.pdf', '34,52'],
                     ['ca6a86753400d6dd6a1b324c5678b7fb.pdf', '18,69'],
                     ['a8380bf795c71caf8185fb11395df138.pdf', '27,38'],
                     ['7fd2ce6b08d086c98158b6f2fa0293b0.pdf', '32,48'],
                     ['dd1adb4dc2014c7abcf403ef15a01eb5.pdf', '2,12'],
                     ['error50.pdf', '1,-1'],
                     ['error59.pdf', '1,-1'],
                     ['error51.pdf', '1,-1'],
                     ['error7.pdf', '39,57'],
                     ]
    index = 1
    # test_one(file_path+test_pdf_list[index][0], page_no_range=test_pdf_list[index][1], from_remote=True)


    # 测试maxcompute模式
    # _process = mp.Process(target=test_maxcompute, args=(file_path, '1,-1',))
    # _process.start()
    # _process.join()