core: libs: commonwealth: general: Add delete_everything_stream

Signed-off-by: Patrick José Pereira <patrickelectric@gmail.com>

Author: patrickelectric

core/libs/commonwealth/commonwealth/utils/general.py

@@ -4,6 +4,7 @@
 from enum import Enum
 from functools import cache
 from pathlib import Path
+from typing import Any, AsyncGenerator
 
 import psutil
 from loguru import logger
@@ -69,6 +70,74 @@ def delete_everything(path: Path) -> None:
             logger.warning(f"Failed to delete: {item}, {exception}")
 
 
+async def delete_everything_stream(path: Path) -> AsyncGenerator[dict[str, Any], None]:
+    """Delete everything in a path and yield information about each file being deleted.
+
+    Args:
+        path: Path to delete
+
+    Yields:
+        Dictionary containing information about each file being deleted:
+        {
+            'path': str,  # Path of the file being deleted
+            'size': int,  # Size of the file in bytes
+            'type': str,  # 'file' or 'directory'
+            'success': bool  # Whether deletion was successful
+        }
+    """
+    if path.is_file() and not file_is_open(path):
+        try:
+            size = path.stat().st_size
+            path.unlink()
+            # fmt: off
+            yield {
+                "path": str(path),
+                "size": size,
+                "type": "file",
+                "success": True
+            }
+            # fmt: on
+        except Exception as exception:
+            logger.warning(f"Failed to delete: {path}, {exception}")
+            # fmt: off
+            yield {
+                "path": str(path),
+                "size": 0,
+                "type": "file",
+                "success": False
+            }
+            # fmt: on
+        return
+
+    for item in path.glob("*"):
+        try:
+            if item.is_file() and not file_is_open(item):
+                size = item.stat().st_size
+                item.unlink()
+                # fmt: off
+                yield {
+                    "path": str(item),
+                    "size": size,
+                    "type": "file",
+                    "success": True
+                }
+                # fmt: on
+            if item.is_dir() and not item.is_symlink():
+                # Delete folder contents
+                async for info in delete_everything_stream(item):
+                    yield info
+        except Exception as exception:
+            logger.warning(f"Failed to delete: {item}, {exception}")
+            # fmt: off
+            yield {
+                "path": str(item),
+                "size": 0,
+                "type": "directory" if item.is_dir() else "file",
+                "success": False
+            }
+            # fmt: on
+
+
 def file_is_open(path: Path) -> bool:
     try:
         kernel_functions_timeout = str(2)