def purge(*, all=False):
"""Manually purge expired vbox entries and remove their folders.
Args:
all (bool): If True, delete all folders, even non-empty ones and those not in _open_boxes.
"""
base_dir = gw.resource(*VBOX_PATH)
with _gc_lock:
now = time.time()
expired = [bid for bid, exp in _open_boxes.items() if exp < now]
known_prefixes = set()
# Clean up known/active expired boxes
for bid in expired:
del _open_boxes[bid]
try:
short, _ = bid.split(".", 1)
known_prefixes.add(short)
folder = os.path.join(base_dir, short)
if os.path.isdir(folder) and (all or not os.listdir(folder)):
shutil.rmtree(folder)
except Exception as e:
gw.error(f"[PURGE] Error cleaning known box {bid}: {e}")
# Clean up orphan folders
for name in os.listdir(base_dir):
if name in known_prefixes:
continue
path = os.path.join(base_dir, name)
if not os.path.isdir(path):
continue
if all:
try:
shutil.rmtree(path)
except Exception as e:
gw.error(f"[PURGE] Error removing orphan box {name}: {e}")
else:
try:
if not os.listdir(path):
shutil.rmtree(path)
except Exception as e:
gw.error(f"[PURGE] Error removing orphan empty box {name}: {e}")