diff --git a/agentuniverse/agent/action/tool/common_tool/write_word_tool.py b/agentuniverse/agent/action/tool/common_tool/write_word_tool.py new file mode 100644 index 00000000..4b7cade2 --- /dev/null +++ b/agentuniverse/agent/action/tool/common_tool/write_word_tool.py @@ -0,0 +1,62 @@ +import os +import json +from typing import Any, Dict + +from agentuniverse.agent.action.tool.tool import Tool + + +class WriteWordDocumentTool(Tool): + def execute(self, file_path: str, content: str = "", append: bool = False) -> str: + directory = os.path.dirname(file_path) + if directory and not os.path.exists(directory): + try: + os.makedirs(directory, exist_ok=True) + except Exception as e: + return json.dumps( + {"error": f"Failed to create directory: {str(e)}", "file_path": file_path, "status": "error"} + ) + + try: + from docx import Document # type: ignore + except ImportError as e: + return json.dumps( + { + "error": f"python-docx is required to write Word documents: {str(e)}", + "file_path": file_path, + "status": "error", + } + ) + + if not file_path.lower().endswith(".docx"): + return json.dumps( + {"error": "The target file must have a .docx extension.", "file_path": file_path, "status": "error"} + ) + + document = None + if append and os.path.exists(file_path): + try: + document = Document(file_path) + except Exception as e: + return json.dumps( + {"error": f"Failed to load existing document: {str(e)}", "file_path": file_path, "status": "error"} + ) + else: + document = Document() + + try: + document.add_paragraph(content) + document.save(file_path) + file_size = os.path.getsize(file_path) + return json.dumps( + { + "file_path": file_path, + "bytes_written": len(content.encode("utf-8")), + "file_size": file_size, + "append_mode": append, + "status": "success", + } + ) + except Exception as e: + return json.dumps( + {"error": f"Failed to write document: {str(e)}", "file_path": file_path, "status": "error"} + ) diff --git a/tests/test_agentuniverse/unit/agent/action/tool/test_write_word_tool.py b/tests/test_agentuniverse/unit/agent/action/tool/test_write_word_tool.py new file mode 100644 index 00000000..c987aadd --- /dev/null +++ b/tests/test_agentuniverse/unit/agent/action/tool/test_write_word_tool.py @@ -0,0 +1,89 @@ +import os +import json +import tempfile +import unittest +from agentuniverse.agent.action.tool.common_tool.write_word_tool import WriteWordDocumentTool + + +class WriteWordDocumentToolTest(unittest.TestCase): + def setUp(self): + self.tool = WriteWordDocumentTool() + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + for root, dirs, files in os.walk(self.temp_dir, topdown=False): + for name in files: + os.unlink(os.path.join(root, name)) + for name in dirs: + os.rmdir(os.path.join(root, name)) + os.rmdir(self.temp_dir) + + def test_write_new_word_file(self): + file_path = os.path.join(self.temp_dir, "test_new.docx") + content = "***This is a test paragraph.***" + + result_json = self.tool.execute(file_path=file_path, content=content, append=False) + result = json.loads(result_json) + + self.assertEqual(result["status"], "success") + self.assertEqual(result["file_path"], file_path) + self.assertTrue(os.path.exists(file_path)) + + def test_append_to_word_file(self): + file_path = os.path.join(self.temp_dir, "test_append.docx") + + initial_content = "Initial paragraph." + self.tool.execute(file_path=file_path, content=initial_content, append=False) + + append_content = "Appended paragraph." + result_json = self.tool.execute(file_path=file_path, content=append_content, append=True) + result = json.loads(result_json) + + self.assertEqual(result["status"], "success") + self.assertEqual(result["append_mode"], True) + + def test_invalid_file_extension(self): + file_path = os.path.join(self.temp_dir, "invalid_file.txt") + content = "This should fail." + + result_json = self.tool.execute(file_path=file_path, content=content, append=False) + result = json.loads(result_json) + + self.assertEqual(result["status"], "error") + self.assertIn("The target file must have a .docx extension.", result["error"]) + + def test_create_directory_structure(self): + file_path = os.path.join(self.temp_dir, "nested/dir/structure/test.docx") + content = "Test content in nested directory." + + result_json = self.tool.execute(file_path=file_path, content=content, append=False) + result = json.loads(result_json) + + self.assertEqual(result["status"], "success") + self.assertTrue(os.path.exists(file_path)) + self.assertTrue(os.path.isdir(os.path.join(self.temp_dir, "nested/dir/structure"))) + + def test_missing_dependency(self): + original_import = __import__ + + def mock_import(name, *args): + if name == "docx": + raise ImportError("No module named 'docx'") + return original_import(name, *args) + + try: + __builtins__["__import__"] = mock_import + file_path = os.path.join(self.temp_dir, "test_missing_dependency.docx") + content = "This should fail due to missing dependency." + + result_json = self.tool.execute(file_path=file_path, content=content, append=False) + result = json.loads(result_json) + + self.assertEqual(result["status"], "error") + self.assertIn("python-docx is required to write Word documents", result["error"]) + finally: + __builtins__["__import__"] = original_import + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file