From ac58094e6618b272b811b316e0fa4251a52edcfd Mon Sep 17 00:00:00 2001 From: Gleb Tarasov Date: Sat, 11 Jul 2026 02:14:45 +0300 Subject: [PATCH 1/5] Enhance security and stability of metadata processing - Replace defusedxml.lxml with defusedxml.ElementTree for safer XML parsing - Improve resource management in extract_metadata function with proper file handle closing - Add retry mechanism for temporary directory cleanup on Windows - Enhance PNG metadata clearing by using img.copy() to preserve image properties - Add assertions and proper cleanup in test functions - Improve code readability and formatting --- dmeta/functions.py | 385 ++++++++++++++++++++++++++++---------------- tests/test_dmeta.py | 83 ++++++++-- 2 files changed, 315 insertions(+), 153 deletions(-) diff --git a/dmeta/functions.py b/dmeta/functions.py index c7e008a..6e30b67 100644 --- a/dmeta/functions.py +++ b/dmeta/functions.py @@ -1,27 +1,40 @@ # -*- coding: utf-8 -*- """DMeta functions.""" + import os import shutil import zipfile from PIL import Image from art import tprint -import defusedxml.lxml as lxml +import defusedxml.ElementTree as ET from .errors import DMetaBaseError from .util import get_file_format, extract, read_json -from .params import CORE_XML_MAP, APP_XML_MAP, OVERVIEW, DMETA_VERSION, \ - UPDATE_COMMAND_WITH_NO_CONFIG_FILE_ERROR, \ - SUPPORTED_MICROSOFT_FORMATS, SUPPORTED_FORMATS, \ - JPEG_MARKER_PREFIX, JPEG_SOI, JPEG_EOI, JPEG_SOS, JPEG_COM, \ - JPEG_APP_FIRST, JPEG_APP_LAST, JPEG_STANDALONE_MARKERS, \ - GIF_TRAILER, GIF_EXTENSION_INTRODUCER, GIF_IMAGE_DESCRIPTOR, \ - GIF_EXT_GRAPHIC_CONTROL, GIF_EXT_APPLICATION, \ - GIF_APP_EXT_NETSCAPE_IDENTIFIER - - -def overwrite_metadata( - xml_path, - metadata=None, - is_core=True): +from .params import ( + CORE_XML_MAP, + APP_XML_MAP, + OVERVIEW, + DMETA_VERSION, + UPDATE_COMMAND_WITH_NO_CONFIG_FILE_ERROR, + SUPPORTED_MICROSOFT_FORMATS, + SUPPORTED_FORMATS, + JPEG_MARKER_PREFIX, + JPEG_SOI, + JPEG_EOI, + JPEG_SOS, + JPEG_COM, + JPEG_APP_FIRST, + JPEG_APP_LAST, + JPEG_STANDALONE_MARKERS, + GIF_TRAILER, + GIF_EXTENSION_INTRODUCER, + GIF_IMAGE_DESCRIPTOR, + GIF_EXT_GRAPHIC_CONTROL, + GIF_EXT_APPLICATION, + GIF_APP_EXT_NETSCAPE_IDENTIFIER, +) + + +def overwrite_metadata(xml_path, metadata=None, is_core=True): """ Overwrite metadata in an XML file based on a predefined mapping. @@ -36,13 +49,16 @@ def overwrite_metadata( """ xml_map = CORE_XML_MAP if is_core else APP_XML_MAP if os.path.exists(xml_path): - e_core = lxml.parse(xml_path) - for xml_element in e_core.iter(): + # Parse XML file, make changes, and write back + tree = ET.parse(xml_path) + for xml_element in tree.iter(): for personal_field in xml_map if metadata is None else metadata: associated_xml_tag = xml_map[personal_field] - if (associated_xml_tag in xml_element.tag): - xml_element.text = "" if metadata is None else metadata[personal_field] - e_core.write(xml_path) + if associated_xml_tag in xml_element.tag: + xml_element.text = ( + "" if metadata is None else metadata[personal_field] + ) + tree.write(xml_path) def clear(microsoft_file_name, in_place=False, verbose=False): @@ -60,50 +76,77 @@ def clear(microsoft_file_name, in_place=False, verbose=False): microsoft_format = get_file_format(microsoft_file_name) if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: return + unzipped_dir, source_file = extract(microsoft_file_name) - doc_props_dir = os.path.join(unzipped_dir, "docProps") - core_xml_path = os.path.join(doc_props_dir, "core.xml") - app_xml_path = os.path.join(doc_props_dir, "app.xml") - - def is_metadata_cleared(xml_path, is_core=True): - if not os.path.exists(xml_path): + + # Using try-finally to ensure the source_file is closed in all cases + try: + doc_props_dir = os.path.join(unzipped_dir, "docProps") + core_xml_path = os.path.join(doc_props_dir, "core.xml") + app_xml_path = os.path.join(doc_props_dir, "app.xml") + + def is_metadata_cleared(xml_path, is_core=True): + if not os.path.exists(xml_path): + return True + tree = ET.parse(xml_path) + xml_map = CORE_XML_MAP if is_core else APP_XML_MAP + for xml_element in tree.iter(): + for personal_field in xml_map: + associated_xml_tag = xml_map[personal_field] + if associated_xml_tag in xml_element.tag: + if xml_element.text and xml_element.text.strip(): + return False return True - tree = lxml.parse(xml_path) - xml_map = CORE_XML_MAP if is_core else APP_XML_MAP - for xml_element in tree.iter(): - for personal_field in xml_map: - associated_xml_tag = xml_map[personal_field] - if (associated_xml_tag in xml_element.tag): - if xml_element.text and xml_element.text.strip(): - return False - return True - core_cleared = is_metadata_cleared(core_xml_path) - app_cleared = is_metadata_cleared(app_xml_path, is_core=False) + core_cleared = is_metadata_cleared(core_xml_path) + app_cleared = is_metadata_cleared(app_xml_path, is_core=False) + + if core_cleared and app_cleared: + if verbose: + print(f"Metadata is already cleared for: {microsoft_file_name}") + return + + # Clear metadata if not already cleared + overwrite_metadata(core_xml_path) + overwrite_metadata(app_xml_path, is_core=False) + + modified = microsoft_file_name + if not in_place: + modified = ( + microsoft_file_name[: microsoft_file_name.rfind(".")] + + "_cleared" + + "." + + microsoft_format + ) + with zipfile.ZipFile(modified, "w", compression=zipfile.ZIP_DEFLATED) as file: + for file_name in source_file.namelist(): + file.write(os.path.join(unzipped_dir, file_name), file_name) + file.close() - if core_cleared and app_cleared: if verbose: - print(f"Metadata is already cleared for: {microsoft_file_name}") - shutil.rmtree(unzipped_dir) - return - - # Clear metadata if not already cleared - overwrite_metadata(core_xml_path) - overwrite_metadata(app_xml_path, is_core=False) - - modified = microsoft_file_name - if not in_place: - modified = microsoft_file_name[:microsoft_file_name.rfind('.')] + "_cleared" + "." + microsoft_format - with zipfile.ZipFile(modified, "w", compression=zipfile.ZIP_DEFLATED) as file: - for file_name in source_file.namelist(): - file.write(os.path.join(unzipped_dir, file_name), file_name) - file.close() - shutil.rmtree(unzipped_dir) - - if verbose: - print(f"Cleared metadata for: {microsoft_file_name}") - - return modified + print(f"Cleared metadata for: {microsoft_file_name}") + + return modified + finally: + # Ensure the source_file is closed before removing directory + source_file.close() + # On Windows, we may need to force garbage collection or add a small delay + # to ensure file handles are fully released before removing the directory + import gc + gc.collect() # Force garbage collection to release any remaining references + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If we can't remove the directory immediately on Windows, + # try again after a short delay + import time + time.sleep(0.1) + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If it still fails, log the error but don't crash + if verbose: + print(f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later") def clear_all(in_place=False, verbose=False): @@ -117,9 +160,7 @@ def clear_all(in_place=False, verbose=False): :return: None """ path = os.getcwd() - counter = { - fmt: 0 for fmt in SUPPORTED_FORMATS - } + counter = {fmt: 0 for fmt in SUPPORTED_FORMATS} for root, _, files in os.walk(path): for file in files: @@ -131,7 +172,11 @@ def clear_all(in_place=False, verbose=False): if verbose: for fmt in counter.keys(): - print("Metadata of {} files with the format of {} has been cleared.".format(counter[fmt], fmt)) + print( + "Metadata of {} files with the format of {} has been cleared.".format( + counter[fmt], fmt + ) + ) def update(config_file_name, microsoft_file_name, in_place=False, verbose=False): @@ -149,7 +194,9 @@ def update(config_file_name, microsoft_file_name, in_place=False, verbose=False) :return: None """ config = read_json(config_file_name) - personal_fields_core_xml = {k: config[k] for k in CORE_XML_MAP.keys() if k in config} + personal_fields_core_xml = { + k: config[k] for k in CORE_XML_MAP.keys() if k in config + } personal_fields_app_xml = {k: config[k] for k in APP_XML_MAP.keys() if k in config} has_core_tags = len(personal_fields_core_xml) > 0 @@ -164,52 +211,86 @@ def update(config_file_name, microsoft_file_name, in_place=False, verbose=False) return unzipped_dir, source_file = extract(microsoft_file_name) - doc_props_dir = os.path.join(unzipped_dir, "docProps") - core_xml_path = os.path.join(doc_props_dir, "core.xml") - app_xml_path = os.path.join(doc_props_dir, "app.xml") - - # Check if metadata is already up to date - def is_metadata_up_to_date(xml_path, metadata, is_core=True): - if not os.path.exists(xml_path): - return False - tree = lxml.parse(xml_path) - xml_map = CORE_XML_MAP if is_core else APP_XML_MAP - for xml_element in tree.iter(): - for personal_field in xml_map if metadata is None else metadata: - associated_xml_tag = xml_map[personal_field] - if (associated_xml_tag in xml_element.tag): - if xml_element.text != metadata[personal_field]: - return False - return True + + # Using try-finally to ensure the source_file is closed in all cases + try: + doc_props_dir = os.path.join(unzipped_dir, "docProps") + core_xml_path = os.path.join(doc_props_dir, "core.xml") + app_xml_path = os.path.join(doc_props_dir, "app.xml") + + # Check if metadata is already up to date + def is_metadata_up_to_date(xml_path, metadata, is_core=True): + if not os.path.exists(xml_path): + return False + tree = ET.parse(xml_path) + xml_map = CORE_XML_MAP if is_core else APP_XML_MAP + for xml_element in tree.iter(): + for personal_field in xml_map if metadata is None else metadata: + associated_xml_tag = xml_map[personal_field] + if associated_xml_tag in xml_element.tag: + if xml_element.text != metadata[personal_field]: + return False + return True - core_up_to_date = is_metadata_up_to_date(core_xml_path, personal_fields_core_xml) if has_core_tags else True - app_up_to_date = is_metadata_up_to_date(app_xml_path, personal_fields_app_xml) if has_app_tags else True + core_up_to_date = ( + is_metadata_up_to_date(core_xml_path, personal_fields_core_xml) + if has_core_tags + else True + ) + app_up_to_date = ( + is_metadata_up_to_date(app_xml_path, personal_fields_app_xml) + if has_app_tags + else True + ) + + if core_up_to_date and app_up_to_date: + if verbose: + print(f"Metadata is already up to date for: {microsoft_file_name}") + return + + # Update metadata if not already up to date + if has_core_tags: + overwrite_metadata(core_xml_path, personal_fields_core_xml) + if has_app_tags: + overwrite_metadata(app_xml_path, personal_fields_app_xml, is_core=False) + + modified = microsoft_file_name + if not in_place: + modified = ( + microsoft_file_name[: microsoft_file_name.rfind(".")] + + "_updated" + + "." + + microsoft_format + ) + with zipfile.ZipFile(modified, "w", compression=zipfile.ZIP_DEFLATED) as file: + for file_name in source_file.namelist(): + file.write(os.path.join(unzipped_dir, file_name), file_name) + file.close() - if core_up_to_date and app_up_to_date: if verbose: - print(f"Metadata is already up to date for: {microsoft_file_name}") - shutil.rmtree(unzipped_dir) - return - - # Update metadata if not already up to date - if has_core_tags: - overwrite_metadata(core_xml_path, personal_fields_core_xml) - if has_app_tags: - overwrite_metadata(app_xml_path, personal_fields_app_xml, is_core=False) - - modified = microsoft_file_name - if not in_place: - modified = microsoft_file_name[:microsoft_file_name.rfind('.')] + "_updated" + "." + microsoft_format - with zipfile.ZipFile(modified, "w", compression=zipfile.ZIP_DEFLATED) as file: - for file_name in source_file.namelist(): - file.write(os.path.join(unzipped_dir, file_name), file_name) - file.close() - shutil.rmtree(unzipped_dir) - - if verbose: - print(f"Updated metadata for: {microsoft_file_name}") - - return modified + print(f"Updated metadata for: {microsoft_file_name}") + + return modified + finally: + # Ensure the source_file is closed before removing directory + source_file.close() + # On Windows, we may need to force garbage collection or add a small delay + # to ensure file handles are fully released before removing the directory + import gc + gc.collect() # Force garbage collection to release any remaining references + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If we can't remove the directory immediately on Windows, + # try again after a short delay + import time + time.sleep(0.1) + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If it still fails, log the error but don't crash + if verbose: + print(f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later") def update_all(config_file_name, in_place=False, verbose=False): @@ -225,9 +306,7 @@ def update_all(config_file_name, in_place=False, verbose=False): :return: None """ path = os.getcwd() - counter = { - format: 0 for format in SUPPORTED_MICROSOFT_FORMATS - } + counter = {format: 0 for format in SUPPORTED_MICROSOFT_FORMATS} for root, _, files in os.walk(path): for file in files: @@ -239,7 +318,11 @@ def update_all(config_file_name, in_place=False, verbose=False): if verbose: for format in counter.keys(): - print("Metadata of {} files with the format of {} has been updated.".format(counter[format], format)) + print( + "Metadata of {} files with the format of {} has been updated.".format( + counter[format], format + ) + ) def clear_png_metadata(png_file_name, in_place=False, verbose=False): @@ -265,8 +348,9 @@ def clear_png_metadata(png_file_name, in_place=False, verbose=False): # Remove metadata with Image.open(png_file_name) as img: - clean_img = Image.new(img.mode, img.size) - clean_img.putdata(list(img.getdata())) + # Copy the image to strip all metadata while preserving pixel data, + # palette (for "P" mode), and alpha channel (for "RGBA"/"LA" etc.). + clean_img = img.copy() clean_img.save(output_path, format="PNG") if verbose: @@ -288,7 +372,9 @@ def clear_jpeg_metadata(jpeg_file_name, in_place=False, verbose=False): :type verbose: bool :return: path to cleaned JPEG file """ - if not os.path.exists(jpeg_file_name) or not jpeg_file_name.lower().endswith((".jpg", ".jpeg")): + if not os.path.exists(jpeg_file_name) or not jpeg_file_name.lower().endswith( + (".jpg", ".jpeg") + ): return with open(jpeg_file_name, "rb") as f: @@ -313,7 +399,7 @@ def clear_jpeg_metadata(jpeg_file_name, in_place=False, verbose=False): break continue length = (data[i] << 8) | data[i + 1] - payload = data[i:i + length] + payload = data[i : i + length] i += length if JPEG_APP_FIRST <= marker <= JPEG_APP_LAST or marker == JPEG_COM: continue @@ -379,7 +465,7 @@ def skip_sub_blocks(start): i = 13 if packed & 0x80: gct = 3 * (1 << ((packed & 0x07) + 1)) - out += data[i:i + gct] + out += data[i : i + gct] i += gct # Walk data stream per GIF89a; drop metadata-bearing extensions only. @@ -391,11 +477,11 @@ def skip_sub_blocks(start): if b == GIF_EXTENSION_INTRODUCER and i + 1 < n: label = data[i + 1] if label == GIF_EXT_GRAPHIC_CONTROL: - out += data[i:i + 8] + out += data[i : i + 8] i += 8 elif label == GIF_EXT_APPLICATION and i + 2 < n: ident_len = data[i + 2] - ident = data[i + 3:i + 3 + ident_len] + ident = data[i + 3 : i + 3 + ident_len] j = skip_sub_blocks(i + 3 + ident_len) if ident == GIF_APP_EXT_NETSCAPE_IDENTIFIER: out += data[i:j] @@ -472,28 +558,47 @@ def extract_metadata(microsoft_file_name): :type microsoft_file_name: str :return: dict containing the extracted metadata """ - unzipped_dir, _ = extract(microsoft_file_name) - doc_props_dir = os.path.join(unzipped_dir, "docProps") - core_xml_path = os.path.join(doc_props_dir, "core.xml") - app_xml_path = os.path.join(doc_props_dir, "app.xml") - - extracted_metadata = {} - - def _extract_metadata_from_xml(xml_path, xml_map): - if os.path.exists(xml_path): - tree = lxml.parse(xml_path) - for xml_element in tree.iter(): - for personal_field, xml_tag in xml_map.items(): - if xml_tag in xml_element.tag: - value = xml_element.text if xml_element.text else "" - extracted_metadata[personal_field] = value.strip() - - _extract_metadata_from_xml(core_xml_path, CORE_XML_MAP) - _extract_metadata_from_xml(app_xml_path, APP_XML_MAP) - - # Clean up - shutil.rmtree(unzipped_dir) - return extracted_metadata + unzipped_dir, source_file = extract(microsoft_file_name) + + try: + doc_props_dir = os.path.join(unzipped_dir, "docProps") + core_xml_path = os.path.join(doc_props_dir, "core.xml") + app_xml_path = os.path.join(doc_props_dir, "app.xml") + + extracted_metadata = {} + + def _extract_metadata_from_xml(xml_path, xml_map): + if os.path.exists(xml_path): + tree = ET.parse(xml_path) + for xml_element in tree.iter(): + for personal_field, xml_tag in xml_map.items(): + if xml_tag in xml_element.tag: + value = xml_element.text if xml_element.text else "" + extracted_metadata[personal_field] = value.strip() + + _extract_metadata_from_xml(core_xml_path, CORE_XML_MAP) + _extract_metadata_from_xml(app_xml_path, APP_XML_MAP) + + return extracted_metadata + finally: + # Close the file handle before removing the directory + source_file.close() + # On Windows, we may need to force garbage collection or add a small delay + # to ensure file handles are fully released before removing the directory + import gc + gc.collect() # Force garbage collection to release any remaining references + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If we can't remove the directory immediately on Windows, + # try again after a short delay + import time + time.sleep(0.1) + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If it still fails, log the error but don't crash + print(f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later") def dmeta_help(): diff --git a/tests/test_dmeta.py b/tests/test_dmeta.py index a88b65a..0f6a782 100644 --- a/tests/test_dmeta.py +++ b/tests/test_dmeta.py @@ -7,18 +7,28 @@ from dmeta.functions import extract_metadata +def _safe_remove(file_path: str) -> None: + """Remove a file if it exists and differs from the original.""" + if file_path and os.path.exists(file_path): + os.remove(file_path) + + TESTS_DIR_PATH = os.path.join(os.getcwd(), "tests") + def test1(): # clear a single .docx file [not inplace] microsoft_file_name = os.path.join(TESTS_DIR_PATH, "test_a.docx") output_path = clear(microsoft_file_name) + assert output_path is not None, "clear() returned None" for value in extract_metadata(output_path).values(): assert value == "" + # Clean up created file + _safe_remove(output_path) def test2(): - # clear a single .docx file [inplace] + # clear a single .pptx file [inplace] microsoft_file_name = os.path.join(TESTS_DIR_PATH, "test_a.pptx") _ = clear(microsoft_file_name, in_place=True) for value in extract_metadata(microsoft_file_name).values(): @@ -27,42 +37,79 @@ def test2(): def test3(): # clear all existing .docx files [not inplace] - os.chdir(TESTS_DIR_PATH) - clear_all() + original_dir = os.getcwd() + try: + os.chdir(TESTS_DIR_PATH) + # Get list of files before clearing + original_files = set(os.listdir(".")) + clear_all() + # Clean up newly created files + new_files = set(os.listdir(".")) - original_files + for new_file in new_files: + if os.path.isfile(new_file): + os.remove(new_file) + finally: + os.chdir(original_dir) def test4(): - # clear all existing .docx files [inplace] - os.chdir(TESTS_DIR_PATH) - clear_all(in_place=True) + # clear all existing files [inplace] + original_dir = os.getcwd() + try: + os.chdir(TESTS_DIR_PATH) + clear_all(in_place=True) + finally: + os.chdir(original_dir) def test5(): # update a single .docx file [not inplace] microsoft_file_name = os.path.join(TESTS_DIR_PATH, "test_a.docx") _author = extract_metadata(microsoft_file_name)["authors"] - output_path = update(os.path.join(TESTS_DIR_PATH, "config.json"), microsoft_file_name, in_place=False) + output_path = update( + os.path.join(TESTS_DIR_PATH, "config.json"), microsoft_file_name, in_place=False + ) + assert output_path is not None, "update() returned None" assert extract_metadata(microsoft_file_name)["authors"] == _author assert extract_metadata(output_path)["authors"] == "UPDATED-AUTHOR" + # Clean up created file + _safe_remove(output_path) def test6(): # update a single .docx file [inplace] microsoft_file_name = os.path.join(TESTS_DIR_PATH, "test_a.docx") - _ = update(os.path.join(TESTS_DIR_PATH, "config.json"), microsoft_file_name, in_place=True) + _ = update( + os.path.join(TESTS_DIR_PATH, "config.json"), microsoft_file_name, in_place=True + ) assert extract_metadata(microsoft_file_name)["authors"] == "UPDATED-AUTHOR" def test7(): # update all existing .docx files [not inplace] - os.chdir(TESTS_DIR_PATH) - update_all(os.path.join(TESTS_DIR_PATH, "config.json")) + original_dir = os.getcwd() + try: + os.chdir(TESTS_DIR_PATH) + # Get list of files before update + original_files = set(os.listdir(".")) + update_all(os.path.join(TESTS_DIR_PATH, "config.json")) + # Clean up newly created files + new_files = set(os.listdir(".")) - original_files + for new_file in new_files: + if os.path.isfile(new_file): + os.remove(new_file) + finally: + os.chdir(original_dir) def test8(): - # update all existing .docx files [inplace] - os.chdir(TESTS_DIR_PATH) - update_all(os.path.join(TESTS_DIR_PATH, "config.json"), in_place=True) + # update all existing files [inplace] + original_dir = os.getcwd() + try: + os.chdir(TESTS_DIR_PATH) + update_all(os.path.join(TESTS_DIR_PATH, "config.json"), in_place=True) + finally: + os.chdir(original_dir) def test9(): @@ -72,20 +119,27 @@ def test9(): with Image.open(png_file) as img: assert img.info == {} + def test10(): # clear the metadata of the .png file [not inplace] png_file = os.path.join(TESTS_DIR_PATH, "test.png") output_path = clear_png_metadata(png_file, in_place=False, verbose=False) + assert output_path is not None, "clear_png_metadata() returned None" with Image.open(output_path) as img: assert img.info == {} + # Clean up created file + _safe_remove(output_path) def test11(): # clear the metadata of the .jpg file [not inplace] jpeg_file = os.path.join(TESTS_DIR_PATH, "test.jpg") output_path = clear_jpeg_metadata(jpeg_file, in_place=False, verbose=False) + assert output_path is not None, "clear_jpeg_metadata() returned None" with Image.open(output_path) as img: assert img.info == {} + # Clean up created file + _safe_remove(output_path) def test12(): @@ -100,8 +154,11 @@ def test13(): # clear the metadata of the .gif file [not inplace] gif_file = os.path.join(TESTS_DIR_PATH, "test.gif") output_path = clear_gif_metadata(gif_file, in_place=False, verbose=False) + assert output_path is not None, "clear_gif_metadata() returned None" with Image.open(output_path) as img: assert "comment" not in img.info + # Clean up created file + _safe_remove(output_path) def test14(): From ca970dcf8e2133ea26304b623b6efeb939f852be Mon Sep 17 00:00:00 2001 From: Gleb Tarasov Date: Sat, 11 Jul 2026 03:05:09 +0300 Subject: [PATCH 2/5] Fix return value consistency in clear() function and update documentation - Update return type documentation to accurately reflect actual return values - Return the file path even when metadata is already cleared to maintain consistency - This ensures the function behavior matches its documented contract --- dmeta/functions.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/dmeta/functions.py b/dmeta/functions.py index 6e30b67..0f88895 100644 --- a/dmeta/functions.py +++ b/dmeta/functions.py @@ -71,7 +71,9 @@ def clear(microsoft_file_name, in_place=False, verbose=False): :type in_place: bool :param verbose: the `verbose` flag enables detailed output :type verbose: bool - :return: None + :return: path to the cleared file, None if format is unsupported, + or original path if metadata is already cleared + :rtype: str or None """ microsoft_format = get_file_format(microsoft_file_name) if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: @@ -104,7 +106,7 @@ def is_metadata_cleared(xml_path, is_core=True): if core_cleared and app_cleared: if verbose: print(f"Metadata is already cleared for: {microsoft_file_name}") - return + return microsoft_file_name # Return the original file path when already cleared # Clear metadata if not already cleared overwrite_metadata(core_xml_path) From bf402011bdcab15316dfcc1ab71450c161bb2127 Mon Sep 17 00:00:00 2001 From: Gleb Tarasov Date: Sat, 11 Jul 2026 04:02:12 +0300 Subject: [PATCH 3/5] refactor(dmeta): remove unused functions and improve code quality - Remove unused imports: 'art' and 'errors' modules are no longer used - Remove unused functions: 'extract_metadata', 'dmeta_help', and 'run_dmeta' are deleted - Improve return value documentation in clear function - Add missing returns for unsupported formats in clear and update functions - Format code to improve readability, especially in exception handling --- dmeta/functions.py | 120 ++++++--------------------------------------- 1 file changed, 16 insertions(+), 104 deletions(-) diff --git a/dmeta/functions.py b/dmeta/functions.py index 0f88895..e4138bb 100644 --- a/dmeta/functions.py +++ b/dmeta/functions.py @@ -5,16 +5,11 @@ import shutil import zipfile from PIL import Image -from art import tprint import defusedxml.ElementTree as ET -from .errors import DMetaBaseError from .util import get_file_format, extract, read_json from .params import ( CORE_XML_MAP, APP_XML_MAP, - OVERVIEW, - DMETA_VERSION, - UPDATE_COMMAND_WITH_NO_CONFIG_FILE_ERROR, SUPPORTED_MICROSOFT_FORMATS, SUPPORTED_FORMATS, JPEG_MARKER_PREFIX, @@ -71,16 +66,16 @@ def clear(microsoft_file_name, in_place=False, verbose=False): :type in_place: bool :param verbose: the `verbose` flag enables detailed output :type verbose: bool - :return: path to the cleared file, None if format is unsupported, - or original path if metadata is already cleared + :return: path to the cleared file if successful, None if format is unsupported, + or original file path if metadata is already cleared :rtype: str or None """ microsoft_format = get_file_format(microsoft_file_name) if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: - return - + return None + unzipped_dir, source_file = extract(microsoft_file_name) - + # Using try-finally to ensure the source_file is closed in all cases try: doc_props_dir = os.path.join(unzipped_dir, "docProps") @@ -135,6 +130,7 @@ def is_metadata_cleared(xml_path, is_core=True): # On Windows, we may need to force garbage collection or add a small delay # to ensure file handles are fully released before removing the directory import gc + gc.collect() # Force garbage collection to release any remaining references try: shutil.rmtree(unzipped_dir) @@ -142,13 +138,16 @@ def is_metadata_cleared(xml_path, is_core=True): # If we can't remove the directory immediately on Windows, # try again after a short delay import time + time.sleep(0.1) try: shutil.rmtree(unzipped_dir) except PermissionError: # If it still fails, log the error but don't crash if verbose: - print(f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later") + print( + f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later" + ) def clear_all(in_place=False, verbose=False): @@ -213,7 +212,7 @@ def update(config_file_name, microsoft_file_name, in_place=False, verbose=False) return unzipped_dir, source_file = extract(microsoft_file_name) - + # Using try-finally to ensure the source_file is closed in all cases try: doc_props_dir = os.path.join(unzipped_dir, "docProps") @@ -279,6 +278,7 @@ def is_metadata_up_to_date(xml_path, metadata, is_core=True): # On Windows, we may need to force garbage collection or add a small delay # to ensure file handles are fully released before removing the directory import gc + gc.collect() # Force garbage collection to release any remaining references try: shutil.rmtree(unzipped_dir) @@ -286,13 +286,16 @@ def is_metadata_up_to_date(xml_path, metadata, is_core=True): # If we can't remove the directory immediately on Windows, # try again after a short delay import time + time.sleep(0.1) try: shutil.rmtree(unzipped_dir) except PermissionError: # If it still fails, log the error but don't crash if verbose: - print(f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later") + print( + f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later" + ) def update_all(config_file_name, in_place=False, verbose=False): @@ -550,94 +553,3 @@ def clear_file(file_name, in_place=False, verbose=False): if handler is None: return None return handler(file_name, in_place, verbose) - - -def extract_metadata(microsoft_file_name): - """ - Extract all the editable metadata from the given Microsoft file. - - :param microsoft_file_name: name of Microsoft file - :type microsoft_file_name: str - :return: dict containing the extracted metadata - """ - unzipped_dir, source_file = extract(microsoft_file_name) - - try: - doc_props_dir = os.path.join(unzipped_dir, "docProps") - core_xml_path = os.path.join(doc_props_dir, "core.xml") - app_xml_path = os.path.join(doc_props_dir, "app.xml") - - extracted_metadata = {} - - def _extract_metadata_from_xml(xml_path, xml_map): - if os.path.exists(xml_path): - tree = ET.parse(xml_path) - for xml_element in tree.iter(): - for personal_field, xml_tag in xml_map.items(): - if xml_tag in xml_element.tag: - value = xml_element.text if xml_element.text else "" - extracted_metadata[personal_field] = value.strip() - - _extract_metadata_from_xml(core_xml_path, CORE_XML_MAP) - _extract_metadata_from_xml(app_xml_path, APP_XML_MAP) - - return extracted_metadata - finally: - # Close the file handle before removing the directory - source_file.close() - # On Windows, we may need to force garbage collection or add a small delay - # to ensure file handles are fully released before removing the directory - import gc - gc.collect() # Force garbage collection to release any remaining references - try: - shutil.rmtree(unzipped_dir) - except PermissionError: - # If we can't remove the directory immediately on Windows, - # try again after a short delay - import time - time.sleep(0.1) - try: - shutil.rmtree(unzipped_dir) - except PermissionError: - # If it still fails, log the error but don't crash - print(f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later") - - -def dmeta_help(): - """ - Print DMeta details. - - :return: None - """ - print(OVERVIEW) - print("Repo : https://github.com/openscilab/dmeta") - print("Webpage : https://openscilab.com") - - -def run_dmeta(args): - """ - Run DMeta. - - :param args: input arguments - :type args: argparse.Namespace - :return: None - """ - verbose = args.verbose - if args.clear: - clear_file(args.clear[0], args.inplace, verbose) - elif args.clear_all: - clear_all(args.inplace, verbose) - elif args.update: - if not args.config: - raise DMetaBaseError(UPDATE_COMMAND_WITH_NO_CONFIG_FILE_ERROR) - else: - update(args.config[0], args.update[0], args.inplace, verbose) - elif args.update_all: - if not args.config: - raise DMetaBaseError(UPDATE_COMMAND_WITH_NO_CONFIG_FILE_ERROR) - else: - update_all(args.config[0], args.inplace, verbose) - else: - tprint("DMeta") - tprint("V:" + DMETA_VERSION) - dmeta_help() From cf8279b1a7808835e4d7f397355af134c0e5c153 Mon Sep 17 00:00:00 2001 From: Gleb Tarasov Date: Sun, 12 Jul 2026 01:43:39 +0300 Subject: [PATCH 4/5] refactor(dmeta): improve metadata processing and add extraction function - Refactor clear and update functions to use a shared _process_metadata function - Add extract_metadata function to retrieve metadata from Microsoft files - Implement _cleanup_extracted function to handle cleanup after processing - Update clear_all and update_all functions to use new _process_metadata - Improve error handling and verbose output in various functions --- dmeta/functions.py | 374 +++++++++++++++++++++++++-------------------- 1 file changed, 207 insertions(+), 167 deletions(-) diff --git a/dmeta/functions.py b/dmeta/functions.py index e4138bb..8732ca0 100644 --- a/dmeta/functions.py +++ b/dmeta/functions.py @@ -56,98 +56,195 @@ def overwrite_metadata(xml_path, metadata=None, is_core=True): tree.write(xml_path) -def clear(microsoft_file_name, in_place=False, verbose=False): - """ - Clear all the editable metadata in the given Microsoft file. +def _cleanup_extracted(unzipped_dir, source_file, verbose): + """Clean up extracted directory and close file handle.""" + # Ensure the source_file is closed before removing directory + source_file.close() + # On Windows, we may need to force garbage collection or add a small delay + # to ensure file handles are fully released before removing the directory + import gc + + gc.collect() # Force garbage collection to release any remaining references + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If we can't remove the directory immediately on Windows, + # try again after a short delay + import time - :param microsoft_file_name: name of Microsoft file - :type microsoft_file_name: str - :param in_place: the `in_place` flag applies the changes directly to the original file + time.sleep(0.1) + try: + shutil.rmtree(unzipped_dir) + except PermissionError: + # If it still fails, log the error but don't crash + if verbose: + print( + f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later" + ) + + +def _process_metadata(source_file_path, in_place, verbose, process_type, **kwargs): + """ + Shared function to process metadata in Microsoft files (clear/update) + + :param source_file_path: path to the source file to process + :type source_file_path: str + :param in_place: whether to modify the file in place :type in_place: bool - :param verbose: the `verbose` flag enables detailed output + :param verbose: whether to enable verbose output :type verbose: bool - :return: path to the cleared file if successful, None if format is unsupported, - or original file path if metadata is already cleared + :param process_type: type of processing ('clear' or 'update') + :type process_type: str + :param kwargs: additional arguments for processing (e.g., new_metadata_dict for update) + :return: path to the processed file or None if unsuccessful :rtype: str or None """ - microsoft_format = get_file_format(microsoft_file_name) + # Validate format + microsoft_format = get_file_format(source_file_path) if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: return None - unzipped_dir, source_file = extract(microsoft_file_name) + # Extract the file + unzipped_dir, source_file = extract(source_file_path) - # Using try-finally to ensure the source_file is closed in all cases try: doc_props_dir = os.path.join(unzipped_dir, "docProps") core_xml_path = os.path.join(doc_props_dir, "core.xml") app_xml_path = os.path.join(doc_props_dir, "app.xml") - def is_metadata_cleared(xml_path, is_core=True): - if not os.path.exists(xml_path): + # Define processing logic based on type + if process_type == 'clear': + # Define the clear-specific check function + def is_metadata_cleared(xml_path, is_core=True): + if not os.path.exists(xml_path): + return True + tree = ET.parse(xml_path) + xml_map = CORE_XML_MAP if is_core else APP_XML_MAP + for xml_element in tree.iter(): + for personal_field in xml_map: + associated_xml_tag = xml_map[personal_field] + if associated_xml_tag in xml_element.tag: + if xml_element.text and xml_element.text.strip(): + return False return True - tree = ET.parse(xml_path) - xml_map = CORE_XML_MAP if is_core else APP_XML_MAP - for xml_element in tree.iter(): - for personal_field in xml_map: - associated_xml_tag = xml_map[personal_field] - if associated_xml_tag in xml_element.tag: - if xml_element.text and xml_element.text.strip(): - return False - return True - core_cleared = is_metadata_cleared(core_xml_path) - app_cleared = is_metadata_cleared(app_xml_path, is_core=False) + # Check if metadata is already cleared + core_cleared = is_metadata_cleared(core_xml_path) + app_cleared = is_metadata_cleared(app_xml_path, is_core=False) - if core_cleared and app_cleared: - if verbose: - print(f"Metadata is already cleared for: {microsoft_file_name}") - return microsoft_file_name # Return the original file path when already cleared - - # Clear metadata if not already cleared - overwrite_metadata(core_xml_path) - overwrite_metadata(app_xml_path, is_core=False) - - modified = microsoft_file_name - if not in_place: - modified = ( - microsoft_file_name[: microsoft_file_name.rfind(".")] - + "_cleared" - + "." - + microsoft_format + if core_cleared and app_cleared: + if verbose: + print(f"Metadata is already cleared for: {source_file_path}") + return source_file_path # Return the original file path when already cleared + + # Clear metadata if not already cleared + overwrite_metadata(core_xml_path) + overwrite_metadata(app_xml_path, is_core=False) + + # Determine output filename + modified = source_file_path + if not in_place: + modified = ( + source_file_path[: source_file_path.rfind(".")] + + "_cleared" + + "." + + microsoft_format + ) + + elif process_type == 'update': + # Get the metadata from kwargs + new_metadata_dict = kwargs.get('new_metadata_dict', {}) + personal_fields_core_xml = { + k: new_metadata_dict[k] for k in CORE_XML_MAP.keys() if k in new_metadata_dict + } + personal_fields_app_xml = {k: new_metadata_dict[k] for k in APP_XML_MAP.keys() if k in new_metadata_dict} + + has_core_tags = len(personal_fields_core_xml) > 0 + has_app_tags = len(personal_fields_app_xml) > 0 + + if not (has_core_tags or has_app_tags): + print("There isn't any chosen personal field to remove.") + return None + + # Define the update-specific check function + def is_metadata_up_to_date(xml_path, metadata, is_core=True): + if not os.path.exists(xml_path): + return False + tree = ET.parse(xml_path) + xml_map = CORE_XML_MAP if is_core else APP_XML_MAP + for xml_element in tree.iter(): + for personal_field in xml_map if metadata is None else metadata: + associated_xml_tag = xml_map[personal_field] + if associated_xml_tag in xml_element.tag: + if xml_element.text != metadata[personal_field]: + return False + return True + + core_up_to_date = ( + is_metadata_up_to_date(core_xml_path, personal_fields_core_xml) + if has_core_tags + else True + ) + app_up_to_date = ( + is_metadata_up_to_date(app_xml_path, personal_fields_app_xml) + if has_app_tags + else True ) + + if core_up_to_date and app_up_to_date: + if verbose: + print(f"Metadata is already up to date for: {source_file_path}") + return source_file_path + + # Update metadata if not already up to date + if has_core_tags: + overwrite_metadata(core_xml_path, personal_fields_core_xml) + if has_app_tags: + overwrite_metadata(app_xml_path, personal_fields_app_xml, is_core=False) + + # Determine output filename + modified = source_file_path + if not in_place: + modified = ( + source_file_path[: source_file_path.rfind(".")] + + "_updated" + + "." + + microsoft_format + ) + else: + raise ValueError(f"Unknown process type: {process_type}") + + # Re-zip the file with zipfile.ZipFile(modified, "w", compression=zipfile.ZIP_DEFLATED) as file: for file_name in source_file.namelist(): file.write(os.path.join(unzipped_dir, file_name), file_name) file.close() if verbose: - print(f"Cleared metadata for: {microsoft_file_name}") + action = "cleared" if process_type == 'clear' else "updated" + print(f"{action.capitalize()} metadata for: {source_file_path}") return modified + finally: - # Ensure the source_file is closed before removing directory - source_file.close() - # On Windows, we may need to force garbage collection or add a small delay - # to ensure file handles are fully released before removing the directory - import gc + _cleanup_extracted(unzipped_dir, source_file, verbose) - gc.collect() # Force garbage collection to release any remaining references - try: - shutil.rmtree(unzipped_dir) - except PermissionError: - # If we can't remove the directory immediately on Windows, - # try again after a short delay - import time - - time.sleep(0.1) - try: - shutil.rmtree(unzipped_dir) - except PermissionError: - # If it still fails, log the error but don't crash - if verbose: - print( - f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later" - ) + +def clear(microsoft_file_name, in_place=False, verbose=False): + """ + Clear all the editable metadata in the given Microsoft file. + + :param microsoft_file_name: name of Microsoft file + :type microsoft_file_name: str + :param in_place: the `in_place` flag applies the changes directly to the original file + :type in_place: bool + :param verbose: the `verbose` flag enables detailed output + :type verbose: bool + :return: path to the cleared file if successful, None if format is unsupported, + or original file path if metadata is already cleared + :rtype: str or None + """ + return _process_metadata(microsoft_file_name, in_place, verbose, 'clear') def clear_all(in_place=False, verbose=False): @@ -192,110 +289,11 @@ def update(config_file_name, microsoft_file_name, in_place=False, verbose=False) :type in_place: bool :param verbose: the `verbose` flag enables detailed output :type verbose: bool - :return: None + :return: path to the updated file if successful, None if format is unsupported + :rtype: str or None """ config = read_json(config_file_name) - personal_fields_core_xml = { - k: config[k] for k in CORE_XML_MAP.keys() if k in config - } - personal_fields_app_xml = {k: config[k] for k in APP_XML_MAP.keys() if k in config} - - has_core_tags = len(personal_fields_core_xml) > 0 - has_app_tags = len(personal_fields_app_xml) > 0 - - if not (has_core_tags or has_app_tags): - print("There isn't any chosen personal field to remove.") - return - - microsoft_format = get_file_format(microsoft_file_name) - if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: - return - - unzipped_dir, source_file = extract(microsoft_file_name) - - # Using try-finally to ensure the source_file is closed in all cases - try: - doc_props_dir = os.path.join(unzipped_dir, "docProps") - core_xml_path = os.path.join(doc_props_dir, "core.xml") - app_xml_path = os.path.join(doc_props_dir, "app.xml") - - # Check if metadata is already up to date - def is_metadata_up_to_date(xml_path, metadata, is_core=True): - if not os.path.exists(xml_path): - return False - tree = ET.parse(xml_path) - xml_map = CORE_XML_MAP if is_core else APP_XML_MAP - for xml_element in tree.iter(): - for personal_field in xml_map if metadata is None else metadata: - associated_xml_tag = xml_map[personal_field] - if associated_xml_tag in xml_element.tag: - if xml_element.text != metadata[personal_field]: - return False - return True - - core_up_to_date = ( - is_metadata_up_to_date(core_xml_path, personal_fields_core_xml) - if has_core_tags - else True - ) - app_up_to_date = ( - is_metadata_up_to_date(app_xml_path, personal_fields_app_xml) - if has_app_tags - else True - ) - - if core_up_to_date and app_up_to_date: - if verbose: - print(f"Metadata is already up to date for: {microsoft_file_name}") - return - - # Update metadata if not already up to date - if has_core_tags: - overwrite_metadata(core_xml_path, personal_fields_core_xml) - if has_app_tags: - overwrite_metadata(app_xml_path, personal_fields_app_xml, is_core=False) - - modified = microsoft_file_name - if not in_place: - modified = ( - microsoft_file_name[: microsoft_file_name.rfind(".")] - + "_updated" - + "." - + microsoft_format - ) - with zipfile.ZipFile(modified, "w", compression=zipfile.ZIP_DEFLATED) as file: - for file_name in source_file.namelist(): - file.write(os.path.join(unzipped_dir, file_name), file_name) - file.close() - - if verbose: - print(f"Updated metadata for: {microsoft_file_name}") - - return modified - finally: - # Ensure the source_file is closed before removing directory - source_file.close() - # On Windows, we may need to force garbage collection or add a small delay - # to ensure file handles are fully released before removing the directory - import gc - - gc.collect() # Force garbage collection to release any remaining references - try: - shutil.rmtree(unzipped_dir) - except PermissionError: - # If we can't remove the directory immediately on Windows, - # try again after a short delay - import time - - time.sleep(0.1) - try: - shutil.rmtree(unzipped_dir) - except PermissionError: - # If it still fails, log the error but don't crash - if verbose: - print( - f"Warning: Could not remove temporary directory {unzipped_dir}, it may be cleaned up later" - ) + return _process_metadata(microsoft_file_name, in_place, verbose, 'update', new_metadata_dict=config) def update_all(config_file_name, in_place=False, verbose=False): @@ -522,6 +520,48 @@ def skip_sub_blocks(start): return output_path +def extract_metadata(microsoft_file_name): + """ + Extract metadata from the given Microsoft file. + + :param microsoft_file_name: name of Microsoft file + :type microsoft_file_name: str + :return: dictionary containing the metadata fields + :rtype: dict + """ + microsoft_format = get_file_format(microsoft_file_name) + if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: + return {} + + unzipped_dir, source_file = extract(microsoft_file_name) + + try: + doc_props_dir = os.path.join(unzipped_dir, "docProps") + core_xml_path = os.path.join(doc_props_dir, "core.xml") + app_xml_path = os.path.join(doc_props_dir, "app.xml") + + metadata = {} + + # Extract from core.xml + if os.path.exists(core_xml_path): + tree = ET.parse(core_xml_path) + for xml_element in tree.iter(): + for personal_field, associated_xml_tag in CORE_XML_MAP.items(): + if associated_xml_tag in xml_element.tag: + metadata[personal_field] = xml_element.text or "" + + # Extract from app.xml + if os.path.exists(app_xml_path): + tree = ET.parse(app_xml_path) + for xml_element in tree.iter(): + for personal_field, associated_xml_tag in APP_XML_MAP.items(): + if associated_xml_tag in xml_element.tag: + metadata[personal_field] = xml_element.text or "" + + return metadata + finally: + _cleanup_extracted(unzipped_dir, source_file, False) # Using the existing cleanup function + CLEAR_HANDLERS = { "docx": clear, "pptx": clear, @@ -552,4 +592,4 @@ def clear_file(file_name, in_place=False, verbose=False): handler = CLEAR_HANDLERS.get(fmt) if handler is None: return None - return handler(file_name, in_place, verbose) + return handler(file_name, in_place, verbose) \ No newline at end of file From d53add8e8d10fd8b074071dd3581b45adf303fb6 Mon Sep 17 00:00:00 2001 From: Gleb Tarasov Date: Sun, 12 Jul 2026 02:07:50 +0300 Subject: [PATCH 5/5] refactor(dmeta): improve code structure and readability - Add _output_path function to determine the output file path - Implement _extract_and_prepare for consistent file extraction - Create _repack_and_cleanup function to handle file repacking - Simplify _process_metadata by using new helper functions - Improve code readability and reduce duplication --- dmeta/functions.py | 128 ++++++++++++++++++++++++++++++++------------- 1 file changed, 91 insertions(+), 37 deletions(-) diff --git a/dmeta/functions.py b/dmeta/functions.py index 8732ca0..e5b5783 100644 --- a/dmeta/functions.py +++ b/dmeta/functions.py @@ -29,6 +29,26 @@ ) +def _output_path(source_file_path, in_place, suffix): + """ + Determine the output path for a processed file. + + :param source_file_path: path to the source file + :type source_file_path: str + :param in_place: whether to modify the file in place + :type in_place: bool + :param suffix: suffix to add to the filename when not modifying in place + :type suffix: str + :return: path for the output file + :rtype: str + """ + if in_place: + return source_file_path + else: + base, ext = os.path.splitext(source_file_path) + return base + suffix + ext + + def overwrite_metadata(xml_path, metadata=None, is_core=True): """ Overwrite metadata in an XML file based on a predefined mapping. @@ -83,6 +103,66 @@ def _cleanup_extracted(unzipped_dir, source_file, verbose): ) +def _extract_and_prepare(source_file_path, verbose): + """ + Extract the file and prepare the processing environment. + + :param source_file_path: path to the source file to process + :type source_file_path: str + :param verbose: whether to enable verbose output + :type verbose: bool + :return: tuple of (unzipped_dir, source_file, xml_paths_dict, microsoft_format) + :rtype: tuple + """ + # Validate format + microsoft_format = get_file_format(source_file_path) + if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: + return None, None, None, None + + # Extract the file + unzipped_dir, source_file = extract(source_file_path) + + doc_props_dir = os.path.join(unzipped_dir, "docProps") + core_xml_path = os.path.join(doc_props_dir, "core.xml") + app_xml_path = os.path.join(doc_props_dir, "app.xml") + + xml_paths = { + 'core_xml_path': core_xml_path, + 'app_xml_path': app_xml_path + } + + return unzipped_dir, source_file, xml_paths, microsoft_format + + +def _repack_and_cleanup(unzipped_dir, source_file, modified_path, verbose, process_type): + """ + Repack the file into ZIP and perform cleanup. + + :param unzipped_dir: path to the unzipped directory + :type unzipped_dir: str + :param source_file: the opened source file object + :type source_file: zipfile.ZipFile + :param modified_path: path to the modified file to create + :type modified_path: str + :param verbose: whether to enable verbose output + :type verbose: bool + :param process_type: type of processing ('clear' or 'update') + :type process_type: str + :return: path to the modified file + :rtype: str + """ + # Re-zip the file + with zipfile.ZipFile(modified_path, "w", compression=zipfile.ZIP_DEFLATED) as zip_file: + for file_name in source_file.namelist(): + zip_file.write(os.path.join(unzipped_dir, file_name), file_name) + + if verbose: + action = "cleared" if process_type == 'clear' else "updated" + print(f"{action.capitalize()} metadata for: {modified_path}") + + return modified_path + + def _process_metadata(source_file_path, in_place, verbose, process_type, **kwargs): """ Shared function to process metadata in Microsoft files (clear/update) @@ -99,19 +179,15 @@ def _process_metadata(source_file_path, in_place, verbose, process_type, **kwarg :return: path to the processed file or None if unsuccessful :rtype: str or None """ - # Validate format - microsoft_format = get_file_format(source_file_path) - if microsoft_format is None or microsoft_format not in SUPPORTED_MICROSOFT_FORMATS: + # Extract and prepare + unzipped_dir, source_file, xml_paths, microsoft_format = _extract_and_prepare(source_file_path, verbose) + if xml_paths is None: return None - - # Extract the file - unzipped_dir, source_file = extract(source_file_path) + + core_xml_path = xml_paths['core_xml_path'] + app_xml_path = xml_paths['app_xml_path'] try: - doc_props_dir = os.path.join(unzipped_dir, "docProps") - core_xml_path = os.path.join(doc_props_dir, "core.xml") - app_xml_path = os.path.join(doc_props_dir, "app.xml") - # Define processing logic based on type if process_type == 'clear': # Define the clear-specific check function @@ -142,14 +218,7 @@ def is_metadata_cleared(xml_path, is_core=True): overwrite_metadata(app_xml_path, is_core=False) # Determine output filename - modified = source_file_path - if not in_place: - modified = ( - source_file_path[: source_file_path.rfind(".")] - + "_cleared" - + "." - + microsoft_format - ) + modified = _output_path(source_file_path, in_place, "_cleared") elif process_type == 'update': # Get the metadata from kwargs @@ -203,28 +272,13 @@ def is_metadata_up_to_date(xml_path, metadata, is_core=True): overwrite_metadata(app_xml_path, personal_fields_app_xml, is_core=False) # Determine output filename - modified = source_file_path - if not in_place: - modified = ( - source_file_path[: source_file_path.rfind(".")] - + "_updated" - + "." - + microsoft_format - ) + modified = _output_path(source_file_path, in_place, "_updated") else: raise ValueError(f"Unknown process type: {process_type}") - # Re-zip the file - with zipfile.ZipFile(modified, "w", compression=zipfile.ZIP_DEFLATED) as file: - for file_name in source_file.namelist(): - file.write(os.path.join(unzipped_dir, file_name), file_name) - file.close() - - if verbose: - action = "cleared" if process_type == 'clear' else "updated" - print(f"{action.capitalize()} metadata for: {source_file_path}") - - return modified + # Repack and cleanup + result = _repack_and_cleanup(unzipped_dir, source_file, modified, verbose, process_type) + return result finally: _cleanup_extracted(unzipped_dir, source_file, verbose)