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