import json import urllib.parse import xml.etree.ElementTree as ET from xml.dom import minidom import sys import os import glob def create_element(parent, tag, attrib=None, text=None): if attrib is None: attrib = {} elem = ET.SubElement(parent, tag, attrib) if text: elem.text = text return elem def convert_har_to_jmx(har_file_path, jmx_file_path): # Đọc file HAR try: with open(har_file_path, 'r', encoding='utf-8') as f: har_data = json.load(f) except Exception as e: print(f"❌ Lỗi đọc file HAR ({har_file_path}): {e}") return entries = har_data.get("log", {}).get("entries", []) # Khởi tạo gốc JMX jmeter_test_plan = ET.Element("jmeterTestPlan", version="1.2", properties="5.0", jmeter="5.6.3") hash_tree_root = ET.SubElement(jmeter_test_plan, "hashTree") # Thêm TestPlan test_plan = create_element(hash_tree_root, "TestPlan", {"guiclass": "TestPlanGui", "testclass": "TestPlan", "testname": "HAR to JMX Test Plan"}) create_element(test_plan, "boolProp", {"name": "TestPlan.functional_mode"}, "false") create_element(test_plan, "boolProp", {"name": "TestPlan.serialize_threadgroups"}, "false") # --- Trích xuất domain mặc định từ API đầu tiên --- default_domain = "" for entry in entries: url_str = entry.get("request", {}).get("url", "") if not url_str.startswith("http") or url_str.endswith((".js", ".css", ".png", ".jpg", ".woff2", ".svg", ".ico", ".gif")): continue if "healthcheck" in url_str.lower(): continue parsed = urllib.parse.urlparse(url_str) default_domain = parsed.hostname or "" break create_element(test_plan, "elementProp", {"name": "TestPlan.user_defined_variables", "elementType": "Arguments"}) create_element(test_plan, "stringProp", {"name": "TestPlan.user_define_classpath"}) hash_tree_tp = ET.SubElement(hash_tree_root, "hashTree") # Thêm ThreadGroup thread_group = create_element(hash_tree_tp, "ThreadGroup", {"guiclass": "ThreadGroupGui", "testclass": "ThreadGroup", "testname": "Thread Group"}) create_element(thread_group, "intProp", {"name": "ThreadGroup.num_threads"}, "1") create_element(thread_group, "intProp", {"name": "ThreadGroup.ramp_time"}, "1") create_element(thread_group, "boolProp", {"name": "ThreadGroup.same_user_on_next_iteration"}, "true") create_element(thread_group, "stringProp", {"name": "ThreadGroup.on_sample_error"}, "continue") elementProp = create_element(thread_group, "elementProp", {"name": "ThreadGroup.main_controller", "elementType": "LoopController", "guiclass": "LoopControlPanel", "testclass": "LoopController"}) create_element(elementProp, "boolProp", {"name": "LoopController.continue_forever"}, "false") create_element(elementProp, "intProp", {"name": "LoopController.loops"}, "1") hash_tree_tg = ET.SubElement(hash_tree_tp, "hashTree") # Thêm HTTP Cookie Manager cookie_manager = create_element(hash_tree_tg, "CookieManager", {"guiclass": "CookiePanel", "testclass": "CookieManager", "testname": "HTTP Cookie Manager"}) create_element(cookie_manager, "collectionProp", {"name": "CookieManager.cookies"}) create_element(cookie_manager, "boolProp", {"name": "CookieManager.clearEachIteration"}, "false") ET.SubElement(hash_tree_tg, "hashTree") # --- THÊM USER DEFINED VARIABLES CONFIG ELEMENT --- udv = create_element(hash_tree_tg, "Arguments", { "guiclass": "ArgumentsPanel", "testclass": "Arguments", "testname": "User Defined Variables" }) udv_col = create_element(udv, "collectionProp", {"name": "Arguments.arguments"}) for var_name, var_value in [("SERVER_NAME", default_domain), ("SERVER_PORT", "443"), ("PROTOCOL", "https")]: arg = create_element(udv_col, "elementProp", {"name": var_name, "elementType": "Argument"}) create_element(arg, "stringProp", {"name": "Argument.name"}, var_name) create_element(arg, "stringProp", {"name": "Argument.value"}, var_value) create_element(arg, "stringProp", {"name": "Argument.metadata"}, "=") ET.SubElement(hash_tree_tg, "hashTree") # --- TẠO GLOBAL HEADER MANAGER VÀ TRÍCH XUẤT TỪ 1 API BẤT KỲ --- global_header_manager = create_element(hash_tree_tg, "HeaderManager", {"guiclass": "HeaderPanel", "testclass": "HeaderManager", "testname": "Global HTTP Header Manager"}) header_col_prop = create_element(global_header_manager, "collectionProp", {"name": "HeaderManager.headers"}) extracted_headers = False for entry in entries: request = entry.get("request", {}) url_str = request.get("url", "") # Bỏ qua request tĩnh if not url_str.startswith("http") or url_str.endswith((".js", ".css", ".png", ".jpg", ".woff2", ".svg", ".ico", ".gif")): continue headers = request.get("headers", []) for header in headers: name = header.get("name") value = header.get("value") name_lower = name.lower() # Lọc bỏ rác if name.startswith(":") or name_lower.startswith("sec-") or name_lower in ["cookie", "content-length", "accept-encoding", "host", "connection", "origin", "referer", "user-agent"]: continue elem_prop = create_element(header_col_prop, "elementProp", {"name": "", "elementType": "Header"}) create_element(elem_prop, "stringProp", {"name": "Header.name"}, name) create_element(elem_prop, "stringProp", {"name": "Header.value"}, value) extracted_headers = True if extracted_headers: break ET.SubElement(hash_tree_tg, "hashTree") # --- TẠO TRANSACTION CONTROLLER --- transaction_controller = create_element(hash_tree_tg, "TransactionController", {"guiclass": "TransactionControllerGui", "testclass": "TransactionController", "testname": "Main API Transaction"}) create_element(transaction_controller, "boolProp", {"name": "TransactionController.includeTimers"}, "false") create_element(transaction_controller, "boolProp", {"name": "TransactionController.parent"}, "true") hash_tree_tc = ET.SubElement(hash_tree_tg, "hashTree") # --- ĐẨY CÁC API VÀO --- api_count = 0 for entry in entries: request = entry.get("request", {}) url_str = request.get("url", "") method = request.get("method", "GET") if not url_str.startswith("http") or url_str.endswith((".js", ".css", ".png", ".jpg", ".woff2", ".svg", ".ico", ".gif")): continue # Bỏ qua các API healthcheck if "healthcheck" in url_str.lower(): continue parsed_url = urllib.parse.urlparse(url_str) domain = parsed_url.hostname port = parsed_url.port if parsed_url.port else ("443" if parsed_url.scheme == "https" else "80") protocol = parsed_url.scheme path = parsed_url.path # Xử lý bỏ _timestamp if parsed_url.query: query_params = urllib.parse.parse_qs(parsed_url.query, keep_blank_values=True) if '_timestamp' in query_params: del query_params['_timestamp'] new_query = urllib.parse.urlencode(query_params, doseq=True) if new_query: path += "?" + new_query # TẠO HTTP SAMPLER (dùng biến chung SERVER_NAME, SERVER_PORT, PROTOCOL) http_sampler = create_element(hash_tree_tc, "HTTPSamplerProxy", {"guiclass": "HttpTestSampleGui", "testclass": "HTTPSamplerProxy", "testname": path.split('/')[-1] or "API Request"}) create_element(http_sampler, "stringProp", {"name": "HTTPSampler.domain"}, "${SERVER_NAME}") create_element(http_sampler, "stringProp", {"name": "HTTPSampler.port"}, "${SERVER_PORT}") create_element(http_sampler, "stringProp", {"name": "HTTPSampler.protocol"}, "${PROTOCOL}") create_element(http_sampler, "stringProp", {"name": "HTTPSampler.path"}, path) create_element(http_sampler, "stringProp", {"name": "HTTPSampler.method"}, method) # VẪN GIỮ ĐOẠN NÀY: Fix lỗi "guicomp is null" args_element = create_element(http_sampler, "elementProp", { "name": "HTTPsampler.Arguments", "elementType": "Arguments", "guiclass": "HTTPArgumentsPanel", "testclass": "Arguments", "testname": "User Defined Variables" }) post_data = request.get("postData", {}).get("text", "") if post_data: create_element(http_sampler, "boolProp", {"name": "HTTPSampler.postBodyRaw"}, "true") col_prop = create_element(args_element, "collectionProp", {"name": "Arguments.arguments"}) arg_elem = create_element(col_prop, "elementProp", {"name": "", "elementType": "HTTPArgument"}) create_element(arg_elem, "boolProp", {"name": "HTTPArgument.always_encode"}, "false") create_element(arg_elem, "stringProp", {"name": "Argument.value"}, post_data) create_element(arg_elem, "stringProp", {"name": "Argument.metadata"}, "=") else: create_element(args_element, "collectionProp", {"name": "Arguments.arguments"}) ET.SubElement(hash_tree_tc, "hashTree") api_count += 1 # --- THÊM LISTENER: Summary Report --- summary_report = create_element(hash_tree_tg, "ResultCollector", { "guiclass": "SummaryReport", "testclass": "ResultCollector", "testname": "Summary Report" }) create_element(summary_report, "boolProp", {"name": "ResultCollector.error_logging"}, "false") create_element(summary_report, "objProp", {}) create_element(summary_report, "stringProp", {"name": "filename"}) ET.SubElement(hash_tree_tg, "hashTree") # --- THÊM LISTENER: View Results Tree --- view_results = create_element(hash_tree_tg, "ResultCollector", { "guiclass": "ViewResultsFullVisualizer", "testclass": "ResultCollector", "testname": "View Results Tree" }) create_element(view_results, "boolProp", {"name": "ResultCollector.error_logging"}, "false") create_element(view_results, "objProp", {}) create_element(view_results, "stringProp", {"name": "filename"}) ET.SubElement(hash_tree_tg, "hashTree") # Format XML đẹp và lưu file xml_string = ET.tostring(jmeter_test_plan, 'utf-8') parsed_xml = minidom.parseString(xml_string) pretty_xml = parsed_xml.toprettyxml(indent=" ") with open(jmx_file_path, "w", encoding="utf-8") as f: f.write(pretty_xml) print(f" ✅ Convert xong: {jmx_file_path} (Gồm {api_count} APIs)") def process_all_files(): current_dir = os.path.dirname(os.path.abspath(__file__)) if '__file__' in globals() else os.getcwd() har_dir = os.path.join(current_dir, "har") jmx_dir = os.path.join(current_dir, "jmx") if not os.path.exists(har_dir): print(f"❌ KHÔNG TÌM THẤY THƯ MỤC 'har'!") print(f"👉 Vui lòng tạo một thư mục tên là 'har' nằm cùng cấp với file script này, và thả các file .har vào đó.") return har_files = glob.glob(os.path.join(har_dir, "*.har")) if len(har_files) == 0: print(f"⚠️ Thư mục 'har' đang trống. Hãy copy các file .har vào đây rồi chạy lại script.") return if not os.path.exists(jmx_dir): os.makedirs(jmx_dir) print(f"📁 Đã tạo thư mục '{jmx_dir}' để chứa kết quả.") print(f"\n🔍 TÌM THẤY {len(har_files)} FILE .HAR. BẮT ĐẦU CHUYỂN ĐỔI:\n" + "-"*50) for har_path in har_files: base_name = os.path.splitext(os.path.basename(har_path))[0] jmx_path = os.path.join(jmx_dir, f"{base_name}.jmx") convert_har_to_jmx(har_path, jmx_path) print("-" * 50 + f"\n🎉 HOÀN TẤT! Toàn bộ file .jmx đã nằm sẵn sàng trong thư mục 'jmx'.\n") if __name__ == "__main__": process_all_files() input("\n⏎ Nhấn Enter để thoát...")