FORMAT_CONVERSION_MAXCOMPUTE/format_convert/convert_test.py

import base64
import json
import os
import random
import sys
import time
from multiprocessing.context import Process

sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/../")
from format_convert.utils import get_platform, request_post, get_md5_from_bytes


def test_one(p, 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": 100}
    if from_remote:
        _url = 'http://121.46.18.113:15010/convert'
        # _url = 'http://192.168.2.103:15010/convert'
        # _url = 'http://172.16.160.65:15010/convert'
        result = json.loads(request_post(_url, data, time_out=10000))
    else:
        print("only support remote!")

    print(_md5)
    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(1):
        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)


if __name__ == '__main__':
    if get_platform() == "Windows":
        # file_path = "C:/Users/Administrator/Desktop/error2.swf"
        # file_path = "D:/BIDI_DOC/比地_文档/2022/Test_Interface/20210609202634853485.xlsx"
        # file_path = "D:/BIDI_DOC/比地_文档/2022/Test_ODPS/1624325845476.pdf"
        file_path = "C:/Users/Administrator/Downloads/1652672734044.jpg"
    else:
        file_path = "test1.doc"
    test_one(file_path, from_remote=True)

    # if get_platform() == "Windows":
    #     # file_path_list = ["D:/BIDI_DOC/比地_文档/2022/Test_Interface/1623328459080.doc",
    #     #                   "D:/BIDI_DOC/比地_文档/2022/Test_Interface/94961e1987d1090e.xls",
    #     #                   "D:/BIDI_DOC/比地_文档/2022/Test_Interface/11111111.rar"]
    #     # file_path_list = ["D:/BIDI_DOC/比地_文档/2022/Test_Interface/1623328459080.doc",
    #     #                   "D:/BIDI_DOC/比地_文档/2022/Test_Interface/94961e1987d1090e.xls"]
    #     # file_path_list = ["D:/BIDI_DOC/比地_文档/2022/Test_Interface/1623423836610.pdf"]
    #     file_path_list = ["C:/Users/Administrator/Downloads/广东中检达元检测技术有限公司.pdf",
    #                       "C:/Users/Administrator/Desktop/error16.jpg"]
    # else:
    #     file_path_list = ["1623423836610.pdf"]
    # start_time = time.time()
    # p_list = []
    # for j in range(3):
    #     p = Process(target=test_duplicate, args=(file_path_list, j, ))
    #     p.start()
    #     p_list.append(p)
    # for p in p_list:
    #     p.join()
    # print("finish", time.time() - start_time)