Coverage for enderchest/place.py: 100%
235 statements
« prev ^ index » next coverage.py v7.10.1, created at 2025-07-30 12:06 +0000
1"""Symlinking functionality"""
3import fnmatch
4import itertools
5import json
6import logging
7import os
8from collections import defaultdict
9from collections.abc import Iterable, Sequence
10from pathlib import Path
12from . import filesystem as fs
13from .inventory import load_ender_chest, load_ender_chest_instances, load_shulker_boxes
14from .loggers import IMPORTANT, INVENTORY_LOGGER, PLACE_LOGGER
15from .prompt import prompt
16from .shulker_box import ShulkerBox
19def place_ender_chest(
20 minecraft_root: Path,
21 keep_broken_links: bool = False,
22 keep_stale_links: bool = False,
23 error_handling: str = "abort",
24 relative: bool = True,
25 rollback: bool = False,
26) -> dict[str, dict[Path, list[str]]]:
27 """Link all instance files and folders to all shulker boxes
29 Parameters
30 ----------
31 minecraft_root : Path
32 The root directory that your minecraft stuff (or, at least, the one
33 that's the parent of your EnderChest folder)
34 keep_broken_links : bool, optional
35 By default, this method will remove any broken links in your instances
36 and servers folders. To disable this behavior, pass in
37 `keep_broken_links=True`.
38 keep_stale_links : bool, optional
39 By default, this method will remove any links into your EnderChest folder
40 that are no longer specified by any shulker box (such as because the
41 instance spec or shulker box configuration changed). To disable this
42 behavior, pass in `keep_stale_links=True`.
43 error_handling : str, optional
44 By default, if a linking failure occurs, this method will terminate
45 immediately (`error_handling=abort`). Alternatively,
46 - pass in `error_handling="ignore"` to continue as if the link failure
47 hadn't occurred
48 - pass in `error_handling="skip"` to abort linking the current instance
49 to the current shulker box but otherwise continue on
50 - pass in `error_handling="skip-instance"` to abort linking the current
51 instance altogether but to otherwise continue on with other instances
52 - pass in `error_handling="skip-shulker-box"` to abort linking to the current
53 shulker box altogether but to otherwise continue on with other boxes
54 - pass in `error_handling="prompt"` to ask what to do on each failure
55 relative : bool, optional
56 By default, links will use relative paths when possible. To use absolute
57 paths instead (see: https://bugs.mojang.com/projects/MC/issues/MC-263046),
58 pass in `relative=False`. See note below.
59 rollback: bool, optional
60 In the future in the event of linking errors passing in `rollback=True`
61 can be used to roll back any changes that have already been applied
62 based on the error-handling method specified.
64 Returns
65 -------
66 dict
67 A record of the placed symlinks, structured as a nested dict:
69 - the top-level keys are the instance names, with the values being a map
70 of the links placed within those instances:
71 - the keys of those mappings are the relative paths of the placed
72 symlinks inside the instance folder
73 - the values are the list of shulker boxes, sorted in ascending
74 priority, into which that symlink was linked (explicitly, the
75 _last_ entry in each list corresponds to the shulker box inside
76 which that link currently points)
78 Notes
79 -----
80 - If one of the files or folders being placed is itself a symlink, relative
81 links will be created as *nested* links (a link pointing to the link),
82 whereas in "absolute" mode (`relative=False`), the link that will be
83 placed will point **directly** to the final target
84 - This can lead to the stale-link cleanup behavior not correctly removing
85 an outdated symlink if the fully resolved target of a link falls outside
86 the EnderChest folder
87 - The generated placement record reflects only the placements performed by
88 _this_ placement operation ("stale" links will never be included)
89 - The generated placements record will include broken links irrespective of
90 the `keep_broken_links` argument
91 - If the placement is aborted (`error_handling="abort"` or "Abort" selected
92 from prompt) then the returned placements will be empty
93 """
94 placements: dict[str, dict[Path, list[str]]] = {}
96 if rollback is not False: # pragma: no cover
97 raise NotImplementedError("Rollbacks are not currently supported")
99 try:
100 host = load_ender_chest(minecraft_root).name
101 except (FileNotFoundError, ValueError) as bad_chest:
102 PLACE_LOGGER.error(
103 "Could not load EnderChest from %s:\n %s", minecraft_root, bad_chest
104 )
105 return {}
107 instances = load_ender_chest_instances(minecraft_root, log_level=logging.DEBUG)
109 shulker_boxes: list[ShulkerBox] = []
111 for shulker_box in load_shulker_boxes(minecraft_root, log_level=logging.DEBUG):
112 if not shulker_box.matches_host(host):
113 PLACE_LOGGER.debug(
114 "%s is not intended for linking to this host (%s)",
115 shulker_box.name,
116 host,
117 )
118 continue
119 shulker_boxes.append(shulker_box)
121 skip_boxes: list[ShulkerBox] = []
123 def handle_error(shulker_box: ShulkerBox | None) -> str:
124 """Centralized error-handling
126 Parameters
127 ----------
128 shulker_box:
129 The current shulker box (in case it needs to be added to the skip list)
131 Returns
132 -------
133 str
134 Instructions on what to do next. Options are:
135 - retry
136 - return
137 - break
138 - continue
139 - pass
140 """
141 if error_handling == "prompt":
142 proceed_how = (
143 prompt(
144 "How would you like to proceed?"
145 "\n[Q]uit; [R]etry; [C]ontinue; skip linking the rest of this:"
146 "\n[I]nstance, [S]hulker box, shulker/instance [M]atch?",
147 suggestion="R",
148 )
149 .lower()
150 .replace(" ", "")
151 .replace("-", "")
152 .replace("_", "")
153 )
154 match proceed_how:
155 case "" | "r":
156 proceed_how = "retry"
157 case "" | "i" | "instance" | "skipinstance":
158 proceed_how = "skip-instance"
159 case "q" | "quit" | "abort" | "exit" | "stop":
160 proceed_how = "abort"
161 case "c" | "continue" | "ignore":
162 proceed_how = "ignore"
163 case "m" | "match" | "skip":
164 proceed_how = "skip"
165 case "s" | "shulker" | "shulkerbox" | "skipshulker":
166 proceed_how = "skip-shulker"
167 case _:
168 PLACE_LOGGER.error("Invalid selection.")
169 return handle_error(shulker_box)
170 else:
171 proceed_how = error_handling
173 match proceed_how:
174 case "retry":
175 return "retry"
176 case "abort" | "stop" | "quit" | "exit":
177 PLACE_LOGGER.error("Aborting")
178 return "return"
179 case "ignore":
180 PLACE_LOGGER.debug("Ignoring")
181 return "pass"
182 case "skip":
183 PLACE_LOGGER.warning("Skipping the rest of this match")
184 return "continue"
185 case "skip-instance":
186 PLACE_LOGGER.warning("Skipping any more linking from this instance")
188 return "break"
189 case "skip-shulker-box" | "skip-shulkerbox" | "skip-shulker":
190 PLACE_LOGGER.warning("Skipping any more linking into this shulker box")
191 if shulker_box:
192 skip_boxes.append(shulker_box)
193 return "continue"
194 case _:
195 raise ValueError(
196 f"Unrecognized error-handling method: {error_handling}"
197 )
199 for instance in instances:
200 instance_root = (minecraft_root / instance.root.expanduser()).expanduser()
201 placements[instance.name] = defaultdict(list)
203 handling: str | None = "retry"
204 while handling == "retry":
205 if instance_root.exists():
206 handling = None
207 break
209 PLACE_LOGGER.error(
210 "No minecraft instance exists at %s",
211 instance_root.expanduser().absolute(),
212 )
213 handling = handle_error(None)
214 if handling is not None:
215 match handling:
216 case "return":
217 return {} # intentionally wipe the cache
218 case "break":
219 break
220 case _: # nothing to link, so might as well skip the rest
221 continue
223 # start by removing all existing symlinks into the EnderChest
224 if not keep_stale_links:
225 for file in instance_root.rglob("*"):
226 if file.is_symlink():
227 if fs.links_into_enderchest(minecraft_root, file):
228 PLACE_LOGGER.debug(
229 "Removing old link: %s -> %s", file, os.readlink(file)
230 )
231 file.unlink()
233 for shulker_box in shulker_boxes:
234 if not shulker_box.matches(instance):
235 continue
236 if shulker_box in skip_boxes:
237 continue
239 box_root = shulker_box.root.expanduser().absolute()
241 PLACE_LOGGER.info("Linking %s to %s", instance.root, shulker_box.name)
243 resources = set(_rglob(box_root, shulker_box.max_link_depth))
245 match_exit = "pass"
246 for link_folder in shulker_box.link_folders:
247 resources -= {box_root / link_folder}
248 resources -= set((box_root / link_folder).rglob("*"))
250 handling = "retry"
251 while handling == "retry":
252 try:
253 link_resource(link_folder, box_root, instance_root, relative)
254 placements[instance.name][Path(link_folder)].append(
255 shulker_box.name
256 )
257 handling = None
258 except OSError:
259 PLACE_LOGGER.error(
260 "Error linking shulker box %s to instance %s:"
261 "\n %s is a non-empty directory",
262 shulker_box.name,
263 instance.name,
264 (instance.root / link_folder),
265 )
266 handling = handle_error(shulker_box)
267 if handling is not None:
268 match handling:
269 case "return":
270 return placements
271 case "break":
272 match_exit = "break"
273 break
274 case "continue":
275 match_exit = "continue"
276 break
277 case "pass":
278 continue # or pass--it's the end of the loop
280 if match_exit not in ("break", "continue"):
281 for resource in resources:
282 resource_path = resource.relative_to(box_root)
283 for pattern in shulker_box.do_not_link:
284 if fnmatch.fnmatchcase(
285 str(resource_path), pattern
286 ) or fnmatch.fnmatchcase(
287 str(resource_path), os.path.join("*", pattern)
288 ):
289 PLACE_LOGGER.debug(
290 "Skipping %s (matches pattern %s)",
291 resource_path,
292 pattern,
293 )
294 break
295 else:
296 handling = "retry"
297 while handling == "retry":
298 try:
299 link_resource(
300 resource_path,
301 box_root,
302 instance_root,
303 relative,
304 )
305 placements[instance.name][resource_path].append(
306 shulker_box.name
307 )
308 handling = None
309 except OSError:
310 PLACE_LOGGER.error(
311 "Error linking shulker box %s to instance %s:"
312 "\n %s already exists",
313 shulker_box.name,
314 instance.name,
315 instance.root / resource_path,
316 )
317 handling = handle_error(shulker_box)
318 if handling is not None:
319 match handling:
320 case "return":
321 return placements
322 case "break":
323 match_exit = "break"
324 break
325 case "continue":
326 match_exit = "continue" # technically does nothing
327 break
328 case "pass":
329 continue # or pass--it's the end of the loop
331 # consider this a "finally"
332 if not keep_broken_links:
333 # we clean up as we go, just in case of a failure
334 for file in instance_root.rglob("*"):
335 if not file.exists():
336 PLACE_LOGGER.debug("Removing broken link: %s", file)
337 file.unlink()
339 if match_exit == "break":
340 break
341 return placements
344def link_resource(
345 resource_path: str | Path,
346 shulker_root: Path,
347 instance_root: Path,
348 relative: bool,
349) -> None:
350 """Create a symlink for the specified resource from an instance's space
351 pointing to the tagged file / folder living inside a shulker box.
353 Parameters
354 ----------
355 resource_path : str or Path
356 Location of the resource relative to the instance's ".minecraft" folder
357 shulker_root : Path
358 The path to the shulker box
359 instance_root : Path
360 The path to the instance's ".minecraft" folder
361 relative : bool
362 If True, the link will be use a relative path if possible. Otherwise,
363 an absolute path will be used, regardless of whether a relative or
364 absolute path was provided.
366 Raises
367 ------
368 OSError
369 If a file or non-empty directory already exists where you're attempting
370 to place the symlink
372 Notes
373 -----
374 - This method will create any folders that do not exist within an instance
375 - This method will overwrite existing symlinks and empty folders
376 but will not overwrite or delete any actual files.
377 """
378 instance_path = (instance_root / resource_path).expanduser().absolute()
379 instance_path.parent.mkdir(parents=True, exist_ok=True)
381 target: str | Path = (shulker_root / resource_path).expanduser().absolute()
382 if relative:
383 target = os.path.relpath(target, instance_path.parent)
384 else:
385 target = target.resolve() # type: ignore
387 if instance_path.is_symlink():
388 # remove previous symlink in this spot
389 instance_path.unlink()
390 PLACE_LOGGER.debug("Removed previous link at %s", instance_path)
391 else:
392 try:
393 os.rmdir(instance_path)
394 PLACE_LOGGER.debug("Removed empty directory at %s", instance_path)
395 except FileNotFoundError:
396 pass # A-OK
398 PLACE_LOGGER.debug("Linking %s to %s", instance_path, target)
399 os.symlink(
400 target,
401 instance_path,
402 target_is_directory=(shulker_root / resource_path).is_dir(),
403 )
406def _rglob(root: Path, max_depth: int) -> Iterable[Path]:
407 """Find all files (and directories* and symlinks) in the path up to the
408 specified depth
410 Parameters
411 ----------
412 root : Path
413 The path to search
414 max_depth : int
415 The maximum number of levels to go
417 Returns
418 -------
419 list-like of paths
420 The files (and directories and symlinks) in the path up to that depth
422 Notes
423 -----
424 - Unlike an actual rglob, this method does not return any directories that
425 are not at the maximum depth
426 - Setting max_depth to 0 (or below) will return all files in the root, but
427 ***be warned*** that because this method follows symlinks, you can very
428 easily find yourself in an infinite loop
429 """
430 top_level = root.iterdir()
431 if max_depth == 1:
432 return top_level
433 return itertools.chain(
434 *(
435 _rglob(path, max_depth - 1) if path.is_dir() else (path,)
436 for path in top_level
437 )
438 )
441def cache_placements(
442 minecraft_root: Path, placements: dict[str, dict[Path, list[str]]]
443) -> None:
444 """Write placement record to file
446 Parameters
447 ----------
448 minecraft_root : Path
449 The root directory that your minecraft stuff (or, at least, the one
450 that's the parent of your EnderChest folder)
451 placements : dict
452 A record of placed links, as generated by `place_ender_chest`
453 """
454 cache_file = fs.place_cache(minecraft_root)
455 cache_file.write_text(
456 json.dumps(
457 {
458 instance_name: {
459 str(resource_path): shulker_boxes
460 for resource_path, shulker_boxes in instance_placements.items()
461 }
462 for instance_name, instance_placements in placements.items()
463 },
464 indent=4,
465 sort_keys=False,
466 )
467 )
468 PLACE_LOGGER.debug("Placement cache written to %s", cache_file)
471def load_placement_cache(minecraft_root: Path) -> dict[str, dict[Path, list[str]]]:
472 """Load the placement cache from file
474 Parameters
475 ----------
476 minecraft_root : Path
477 The root directory that your minecraft stuff (or, at least, the one
478 that's the parent of your EnderChest folder)
480 Returns
481 -------
482 dict
483 A record of the placed symlinks, structured as a nested dict, matching
484 the schema of one generated by `place_ender_chest`
486 Raises
487 ------
488 OSError
489 If the placement cache could not be found, read or parsed
490 """
491 try:
492 cache_file = fs.place_cache(minecraft_root)
493 INVENTORY_LOGGER.debug(
494 "Loading placement cache from %s", fs.place_cache(minecraft_root)
495 )
496 raw_dict: dict[str, dict[str, list[str]]] = json.loads(
497 cache_file.read_text("UTF-8")
498 )
499 except json.JSONDecodeError as decode_error:
500 raise OSError(
501 f"{fs.place_cache(minecraft_root)} is corrupted and could not be parsed:"
502 ) from decode_error
503 return {
504 instance_name: {
505 Path(resource_path): shulker_boxes
506 for resource_path, shulker_boxes in instance_placements.items()
507 }
508 for instance_name, instance_placements in raw_dict.items()
509 }
512def trace_resource(
513 minecraft_root: Path,
514 pattern: str,
515 placements: dict[str, dict[Path, list[str]]],
516 instance_name: str | None = None,
517) -> list[tuple[Path, Path, list[str]]]:
518 """Given a filename or glob pattern, return a list of all matching
519 EnderChest-placed symlinks, together with a trace-back of the shulker boxes
520 each link targets
522 Parameters
523 ----------
524 minecraft_root : Path
525 The root directory that your minecraft stuff (or, at least, the one
526 that's the parent of your EnderChest folder)
527 pattern : filename, path or glob pattern
528 The resource to trace
529 placements : dict
530 A record of placed symlinks, such as the one generated by `place_ender_chest`.
531 instance_name : str, optional
532 The name of the instance to search. This variable is case-sensitive.
533 If None is given, all instances will be searched.
535 Returns
536 -------
537 list of (Path, Path, list) tuples
538 - The first item in each list is the path of the instance root
539 - The second item in each list is the path to a linked resource
540 matching the provided pattern (and instance), relative to the instance
541 root
542 - The third item is the list of shulker boxes, sorted in ascending
543 priority, into which that symlink was linked (explicitly, the
544 _last_ entry in each list corresponds to the shulker box inside
545 which that link currently points)
547 Raises
548 ------
549 OSError
550 If no placement cache was provided and the placement cache file could
551 not be found, read or parsed
552 KeyError
553 If there is no instance registered to this EnderChest with the specified
554 name
555 """
556 instances = {
557 instance.name: instance
558 for instance in load_ender_chest_instances(
559 minecraft_root, log_level=logging.DEBUG
560 )
561 }
562 if instance_name is None:
563 return sum(
564 (
565 trace_resource(minecraft_root, pattern, placements, name)
566 for name in instances
567 ),
568 [],
569 )
570 instance_root = instances[instance_name].root
571 matches: list[tuple[Path, Path, list[str]]] = []
572 for resource_path, target_boxes in placements[instance_name].items():
573 if (
574 fnmatch.fnmatchcase(str(resource_path), pattern)
575 or fnmatch.fnmatchcase(str(resource_path), os.path.join("*", pattern))
576 or fnmatch.fnmatchcase(
577 os.path.abspath(minecraft_root / instance_root / resource_path),
578 os.path.join("*", pattern),
579 )
580 ):
581 matches.append((instance_root, resource_path, target_boxes))
582 return matches
585def report_resource_trace(
586 minecraft_root: Path, instance_root: Path, resource_path: Path, boxes: Sequence[str]
587) -> None:
588 """Print (log) the shulker boxes an instance resource is linked to
590 Parameters
591 ----------
592 minecraft_root : Path
593 The root directory that your minecraft stuff (or, at least, the one
594 that's the parent of your EnderChest folder)
595 instance_root : Path
596 The path of the EnderChest-placed symlink
597 resource_path : Path
598 The path to the symlink, relative to the instance root
599 boxes : list of str
600 The names of the shulker boxes, sorted by ascending priority, that are
601 targeted by this symlink (technically only the last entry in this list
602 is the actual target)
603 """
604 symlink_location = instance_root / resource_path
605 if len(boxes) == 0: # pragma: no cover
606 # Since defaultdicts are involved, this could happen accidentally at
607 # some point and should just be ignored
608 return
609 *other_box_names, primary_box_name = boxes
610 try:
611 INVENTORY_LOGGER.log(
612 IMPORTANT,
613 "%s currently resolves to %s",
614 symlink_location,
615 os.path.abspath(
616 (
617 symlink_location / (minecraft_root / symlink_location).readlink()
618 ).expanduser()
619 ),
620 )
621 except OSError:
622 INVENTORY_LOGGER.warning(
623 "%s no longer exists or is not a symlink", symlink_location
624 )
626 INVENTORY_LOGGER.log(
627 IMPORTANT,
628 " based on being linked into shulker box: %s",
629 primary_box_name,
630 )
631 INVENTORY_LOGGER.debug(
632 " - > %s",
633 fs.shulker_box_root(minecraft_root, primary_box_name) / resource_path,
634 )
636 for box_name in reversed(other_box_names):
637 INVENTORY_LOGGER.info(
638 " which overwrote the link into shulker box: %s", box_name
639 )
640 INVENTORY_LOGGER.debug(
641 " - > %s",
642 fs.shulker_box_root(minecraft_root, box_name) / resource_path,
643 )
646def list_placements(
647 minecraft_root: Path, pattern: str, instance_name: str | None = None
648) -> None:
649 """Report all shulker boxes that provide files matching the given pattern
651 Parameters
652 ----------
653 minecraft_root : Path
654 The root directory that your minecraft stuff (or, at least, the one
655 that's the parent of your EnderChest folder)
656 pattern : filename, path or glob pattern
657 The pattern of the resource to trace
658 instance_name : str, optional
659 The name of the instance to search. This variable is case-sensitive.
660 If None is given, all instances will be searched.
661 """
662 try:
663 placements = load_placement_cache(minecraft_root)
664 except OSError as no_cache:
665 INVENTORY_LOGGER.error(
666 "The placement cache could not be loaded:"
667 "\n %s"
668 "\nPlease run enderchest place again to regenerate the cache.",
669 no_cache,
670 )
671 return
672 try:
673 matches = trace_resource(
674 minecraft_root, pattern, placements, instance_name=instance_name
675 )
676 except KeyError:
677 INVENTORY_LOGGER.error(
678 "No instance named %s is registered to this EnderChest", instance_name
679 )
680 return
681 if len(matches) == 0:
682 INVENTORY_LOGGER.warning(
683 "Could not find any placed resources matching the pattern %s%s."
684 "\n\nNote: this command does not check inside linked folders.",
685 pattern,
686 f"\nin the instance {instance_name}" if instance_name else "",
687 )
688 return
689 for match in matches:
690 report_resource_trace(minecraft_root, *match)