fix: (#2) caches now track deleted files

Author: BPR02

beet_observer/data_pack.py

@@ -1,3 +1,4 @@
+from pathlib import PosixPath
 from typing import Any
 
 import beet.contrib.worldgen as wg
@@ -6,9 +7,10 @@
 
 def gen_dp_overlays(
     ctx: Context, ctx_overlay: Context, overlay_dir: str, ignore: list[str]
-) -> None:
+) -> list[PosixPath]:
     """
-    Generates overlays between two datapacks.
+    Generates overlays between two datapacks. \n
+    Returns a list of deleted files from the source pack.
 
     Keyword arguments:  \n
     `ctx` -- the build context  \n
@@ -108,8 +110,9 @@ def gen_dp_overlays(
         ),
     ]
     # for each file type, check for required overlays
+    deleted: list[PosixPath] = []
     for registry, registry_overlay in file_types:
-        check_registry(ctx, overlay_dir, registry, registry_overlay)
+        deleted.extend(check_registry(ctx, overlay_dir, registry, registry_overlay))
 
     # get pack.mcmeta overlay entries
     mcmeta: dict[str, dict[str, list[dict[str, Any]]]] = ctx.data.mcmeta.data.copy()
@@ -154,29 +157,36 @@ def gen_dp_overlays(
     if len(entries) > 0:
         ctx.data.mcmeta.data.update({"overlays": {"entries": entries}})
 
+    return deleted
+
 
 def check_registry(
     ctx: Context,
     overlay_dir: str,
     registry: NamespaceProxy[Any],
     registry_overlay: NamespaceProxy[Any],
-) -> None:
+) -> list[PosixPath]:
     """
-    Generates overlays for each namespace proxy.
+    Generates overlays for each namespace proxy. \n
+    Returns a list of deleted files from the source pack.
 
     Keyword arguments:  \n
     `ctx` -- the build context  \n
     `overlay_dir` -- the directory of the overlay  \n
     `registry` -- the namespace proxy from the build context  \n
     `registry_overlay` -- the namespace proxy from the overlay context  \n
     """
+    # prepare cache deletion list
+    deleted: list[PosixPath] = []
+
     # check each file in the build pack
     for name in list(registry):
         if name in registry_overlay:
             # exists in both, so check if an overlay is needed
             gen_registry_overlay(ctx, overlay_dir, name, registry, registry_overlay)
         else:
             # exists only in overlay, so create a deletion overlay
+            deleted.append(registry[name].source_path)
             gen_registry_overlay(
                 ctx, overlay_dir, name, registry, registry_overlay, "deletion"
             )
@@ -187,6 +197,8 @@ def check_registry(
             ctx, overlay_dir, name, registry, registry_overlay, "addition"
         )
 
+    return deleted
+
 
 def gen_registry_overlay(
     ctx: Context,

beet_observer/plugin.py

@@ -1,3 +1,5 @@
+from pathlib import PosixPath
+
 from beet import Context, run_beet
 
 from .data_pack import gen_dp_overlays
@@ -17,12 +19,52 @@ def beet_default(ctx: Context):
     if ctx.data:
         dp_path = cache.get_path(f"{ctx.directory} saved_data_pack")
         if dp_path.is_dir():
-            ctx.data.load(f"{dp_path}")
+            # get files that were moved to an overlay
+            with open(f"{dp_path}/deleted.txt", mode="r") as del_list:
+                deleted = del_list.read().splitlines()
+
+            # delete files that were moved to an overlay
+            for target in deleted:
+                # get file location
+                target_path = PosixPath(target)
+                folders = target_path.parts
+                ext = "." + folders[-1].split(".")[-1]
+                loc = f"{folders[1]}:{folders[-1].removesuffix(ext)}"
+                # get resource location
+                for location, resource in ctx.data.all(loc):
+                    p = str(resource.source_path)  # type: ignore
+                    resource_path = PosixPath((p[p.find("/data/") + 1 :]))
+                    # delete resource from pack
+                    if target_path == resource_path:
+                        del ctx.data[type(resource)][location]
+
+            # add overlays to pack
+            ctx.data.load(f"{dp_path}/pack")
             cached_dp = True
     if ctx.assets:
         rp_path = cache.get_path(f"{ctx.directory} saved_resource_pack")
         if rp_path.is_dir():
-            ctx.assets.load(f"{rp_path}")
+            # get files that were moved to an overlay
+            with open(f"{rp_path}/deleted.txt", mode="r") as del_list:
+                deleted = del_list.read().splitlines()
+
+            # delete files that were moved to an overlay
+            for target in deleted:
+                # get file location
+                target_path = PosixPath(target)
+                folders = target_path.parts
+                ext = "." + folders[-1].split(".")[-1]
+                loc = f"{folders[1]}:{folders[-1].removesuffix(ext)}"
+                # get resource location
+                for location, resource in ctx.assets.all(loc):
+                    p = str(resource.source_path)  # type: ignore
+                    resource_path = PosixPath((p[p.find("/assets/") + 1 :]))
+                    # delete resource from pack
+                    if target_path == resource_path:
+                        del ctx.assets[type(resource)][location]
+
+            # add overlays to pack
+            ctx.assets.load(f"{rp_path}/pack")
             cached_rp = True
     if cached_dp and cached_rp:
         return
@@ -52,6 +94,8 @@ def beet_default(ctx: Context):
     for overlay in ctx.assets.overlays:
         save_rp.append(overlay)
     # loop through all overlays
+    deleted_dp: list[PosixPath] = []
+    deleted_rp: list[PosixPath] = []
     for overlay in ctx.meta["observer"]["overlays"]:
         # get pack
         if overlay["process"].startswith("https://"):
@@ -70,12 +114,18 @@ def beet_default(ctx: Context):
                 rp_dir = overlay["directory"]
             # compare build pack and overlay pack
             if not cached_dp and ctx.data:
-                gen_dp_overlays(ctx, ctx_overlay, dp_dir, save_dp)
+                deleted_dp = gen_dp_overlays(ctx, ctx_overlay, dp_dir, save_dp)
             if not cached_rp and ctx.assets:
-                gen_rp_overlays(ctx, ctx_overlay, rp_dir, save_rp)
+                deleted_rp = gen_rp_overlays(ctx, ctx_overlay, rp_dir, save_rp)
 
     # save to cache
     if not cached_dp and ctx.data:
-        ctx.data.save(path=dp_path)
+        ctx.data.save(path=f"{dp_path}/pack")
+        with open(f"{dp_path}/deleted.txt", mode="x") as file:
+            for s in deleted_dp:
+                file.write(str(s)[str(s).find("/data/") + 1 :] + "\n")
     if not cached_rp and ctx.assets:
-        ctx.assets.save(path=rp_path)
+        ctx.assets.save(path=f"{rp_path}/pack")
+        with open(f"{rp_path}/deleted.txt", mode="x") as file:
+            for s in deleted_rp:
+                file.write(str(s)[str(s).find("/assets/") + 1 :] + "\n")

beet_observer/resource_pack.py

@@ -1,11 +1,12 @@
+from pathlib import PosixPath
 from typing import Any
 
 from beet import Context, NamespaceProxy
 
 
 def gen_rp_overlays(
     ctx: Context, ctx_overlay: Context, overlay_dir: str, ignore: list[str]
-) -> None:
+) -> list[PosixPath]:
     """
     Generates overlays between two resource packs.
 
@@ -36,8 +37,9 @@ def gen_rp_overlays(
         (ctx.assets.atlases, ctx_overlay.assets.atlases),
     ]
     # for each file type, check for required overlays
+    deleted: list[PosixPath] = []
     for registry, registry_overlay in file_types:
-        check_registry(ctx, overlay_dir, registry, registry_overlay)
+        deleted.extend(check_registry(ctx, overlay_dir, registry, registry_overlay))
 
     # get pack.mcmeta overlay entries
     mcmeta: dict[str, dict[str, list[dict[str, Any]]]] = ctx.assets.mcmeta.data.copy()
@@ -82,13 +84,15 @@ def gen_rp_overlays(
     if len(entries) > 0:
         ctx.assets.mcmeta.data.update({"overlays": {"entries": entries}})
 
+    return deleted
+
 
 def check_registry(
     ctx: Context,
     overlay_dir: str,
     registry: NamespaceProxy[Any],
     registry_overlay: NamespaceProxy[Any],
-) -> None:
+) -> list[PosixPath]:
     """
     Generates overlays for each namespace proxy.
 
@@ -98,13 +102,17 @@ def check_registry(
     `registry` -- the namespace proxy from the build context  \n
     `registry_overlay` -- the namespace proxy from the overlay context  \n
     """
+    # prepare cache deletion list
+    deleted: list[PosixPath] = []
+
     # check each file in the build pack
     for name in list(registry):
         if name in registry_overlay:
             # exists in both, so check if an overlay is needed
             gen_registry_overlay(ctx, overlay_dir, name, registry, registry_overlay)
         else:
             # exists only in overlay, so create a deletion overlay
+            deleted.append(registry[name].source_path)
             gen_registry_overlay(
                 ctx, overlay_dir, name, registry, registry_overlay, "deletion"
             )
@@ -115,6 +123,8 @@ def check_registry(
             ctx, overlay_dir, name, registry, registry_overlay, "addition"
         )
 
+    return deleted
+
 
 def gen_registry_overlay(
     ctx: Context,