Coverage for enderchest/place.py: 100%
235 statements
« prev ^ index » next coverage.py v7.5.0, created at 2024-05-04 01:41 +0000
1"""Symlinking functionality"""
3import fnmatch
4import itertools
5import json
6import logging
7import os
8from collections import defaultdict
9from pathlib import Path
10from typing import Iterable, Sequence
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 f"Could not load EnderChest from {minecraft_root}:\n {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 f"{shulker_box.name} is not intended for linking to this host ({host})"
115 )
116 continue
117 shulker_boxes.append(shulker_box)
119 skip_boxes: list[ShulkerBox] = []
121 def handle_error(shulker_box: ShulkerBox | None) -> str:
122 """Centralized error-handling
124 Parameters
125 ----------
126 shulker_box:
127 The current shulker box (in case it needs to be added to the skip list)
129 Returns
130 -------
131 str
132 Instructions on what to do next. Options are:
133 - retry
134 - return
135 - break
136 - continue
137 - pass
138 """
139 if error_handling == "prompt":
140 proceed_how = (
141 prompt(
142 "How would you like to proceed?"
143 "\n[Q]uit; [R]etry; [C]ontinue; skip linking the rest of this:"
144 "\n[I]nstance, [S]hulker box, shulker/instance [M]atch?",
145 suggestion="R",
146 )
147 .lower()
148 .replace(" ", "")
149 .replace("-", "")
150 .replace("_", "")
151 )
152 match proceed_how:
153 case "" | "r":
154 proceed_how = "retry"
155 case "" | "i" | "instance" | "skipinstance":
156 proceed_how = "skip-instance"
157 case "q" | "quit" | "abort" | "exit" | "stop":
158 proceed_how = "abort"
159 case "c" | "continue" | "ignore":
160 proceed_how = "ignore"
161 case "m" | "match" | "skip":
162 proceed_how = "skip"
163 case "s" | "shulker" | "shulkerbox" | "skipshulker":
164 proceed_how = "skip-shulker"
165 case _:
166 PLACE_LOGGER.error("Invalid selection.")
167 return handle_error(shulker_box)
168 else:
169 proceed_how = error_handling
171 match proceed_how:
172 case "retry":
173 return "retry"
174 case "abort" | "stop" | "quit" | "exit":
175 PLACE_LOGGER.error("Aborting")
176 return "return"
177 case "ignore":
178 PLACE_LOGGER.debug("Ignoring")
179 return "pass"
180 case "skip":
181 PLACE_LOGGER.warning("Skipping the rest of this match")
182 return "continue"
183 case "skip-instance":
184 PLACE_LOGGER.warning("Skipping any more linking from this instance")
186 return "break"
187 case "skip-shulker-box" | "skip-shulkerbox" | "skip-shulker":
188 PLACE_LOGGER.warning("Skipping any more linking into this shulker box")
189 if shulker_box:
190 skip_boxes.append(shulker_box)
191 return "continue"
192 case _:
193 raise ValueError(
194 f"Unrecognized error-handling method: {error_handling}"
195 )
197 for instance in instances:
198 instance_root = (minecraft_root / instance.root.expanduser()).expanduser()
199 placements[instance.name] = defaultdict(list)
201 handling: str | None = "retry"
202 while handling == "retry":
203 if instance_root.exists():
204 handling = None
205 break
207 PLACE_LOGGER.error(
208 "No minecraft instance exists at"
209 f" {instance_root.expanduser().absolute()}"
210 )
211 handling = handle_error(None)
212 if handling is not None:
213 match handling:
214 case "return":
215 return {} # intentionally wipe the cache
216 case "break":
217 break
218 case _: # nothing to link, so might as well skip the rest
219 continue
221 # start by removing all existing symlinks into the EnderChest
222 if not keep_stale_links:
223 for file in instance_root.rglob("*"):
224 if file.is_symlink():
225 if fs.links_into_enderchest(minecraft_root, file):
226 PLACE_LOGGER.debug(
227 f"Removing old link: {file} -> {os.readlink(file)}"
228 )
229 file.unlink()
231 for shulker_box in shulker_boxes:
232 if not shulker_box.matches(instance):
233 continue
234 if shulker_box in skip_boxes:
235 continue
237 box_root = shulker_box.root.expanduser().absolute()
239 PLACE_LOGGER.info(f"Linking {instance.root} to {shulker_box.name}")
241 resources = set(_rglob(box_root, shulker_box.max_link_depth))
243 match_exit = "pass"
244 for link_folder in shulker_box.link_folders:
245 resources -= {box_root / link_folder}
246 resources -= set((box_root / link_folder).rglob("*"))
248 handling = "retry"
249 while handling == "retry":
250 try:
251 link_resource(link_folder, box_root, instance_root, relative)
252 placements[instance.name][Path(link_folder)].append(
253 shulker_box.name
254 )
255 handling = None
256 except OSError:
257 PLACE_LOGGER.error(
258 f"Error linking shulker box {shulker_box.name}"
259 f" to instance {instance.name}:"
260 f"\n {(instance.root / link_folder)} is a"
261 " non-empty directory"
262 )
263 handling = handle_error(shulker_box)
264 if handling is not None:
265 match handling:
266 case "return":
267 return placements
268 case "break":
269 match_exit = "break"
270 break
271 case "continue":
272 match_exit = "continue"
273 break
274 case "pass":
275 continue # or pass--it's the end of the loop
277 if match_exit not in ("break", "continue"):
278 for resource in resources:
279 resource_path = resource.relative_to(box_root)
280 for pattern in shulker_box.do_not_link:
281 if fnmatch.fnmatchcase(
282 str(resource_path), pattern
283 ) or fnmatch.fnmatchcase(
284 str(resource_path), os.path.join("*", pattern)
285 ):
286 PLACE_LOGGER.debug(
287 "Skipping %s (matches pattern %s)",
288 resource_path,
289 pattern,
290 )
291 break
292 else:
293 handling = "retry"
294 while handling == "retry":
295 try:
296 link_resource(
297 resource_path,
298 box_root,
299 instance_root,
300 relative,
301 )
302 placements[instance.name][resource_path].append(
303 shulker_box.name
304 )
305 handling = None
306 except OSError:
307 PLACE_LOGGER.error(
308 f"Error linking shulker box {shulker_box.name}"
309 f" to instance {instance.name}:"
310 f"\n {(instance.root / resource_path)}"
311 " already exists"
312 )
313 handling = handle_error(shulker_box)
314 if handling is not None:
315 match handling:
316 case "return":
317 return placements
318 case "break":
319 match_exit = "break"
320 break
321 case "continue":
322 match_exit = "continue" # technically does nothing
323 break
324 case "pass":
325 continue # or pass--it's the end of the loop
327 # consider this a "finally"
328 if not keep_broken_links:
329 # we clean up as we go, just in case of a failure
330 for file in instance_root.rglob("*"):
331 if not file.exists():
332 PLACE_LOGGER.debug(f"Removing broken link: {file}")
333 file.unlink()
335 if match_exit == "break":
336 break
337 return placements
340def link_resource(
341 resource_path: str | Path,
342 shulker_root: Path,
343 instance_root: Path,
344 relative: bool,
345) -> None:
346 """Create a symlink for the specified resource from an instance's space
347 pointing to the tagged file / folder living inside a shulker box.
349 Parameters
350 ----------
351 resource_path : str or Path
352 Location of the resource relative to the instance's ".minecraft" folder
353 shulker_root : Path
354 The path to the shulker box
355 instance_root : Path
356 The path to the instance's ".minecraft" folder
357 relative : bool
358 If True, the link will be use a relative path if possible. Otherwise,
359 an absolute path will be used, regardless of whether a relative or
360 absolute path was provided.
362 Raises
363 ------
364 OSError
365 If a file or non-empty directory already exists where you're attempting
366 to place the symlink
368 Notes
369 -----
370 - This method will create any folders that do not exist within an instance
371 - This method will overwrite existing symlinks and empty folders
372 but will not overwrite or delete any actual files.
373 """
374 instance_path = (instance_root / resource_path).expanduser().absolute()
375 instance_path.parent.mkdir(parents=True, exist_ok=True)
377 target: str | Path = (shulker_root / resource_path).expanduser().absolute()
378 if relative:
379 target = os.path.relpath(target, instance_path.parent)
380 else:
381 target = target.resolve() # type: ignore
383 if instance_path.is_symlink():
384 # remove previous symlink in this spot
385 instance_path.unlink()
386 PLACE_LOGGER.debug("Removed previous link at %s", instance_path)
387 else:
388 try:
389 os.rmdir(instance_path)
390 PLACE_LOGGER.debug("Removed empty directory at %s", instance_path)
391 except FileNotFoundError:
392 pass # A-OK
394 PLACE_LOGGER.debug("Linking %s to %s", instance_path, target)
395 os.symlink(
396 target,
397 instance_path,
398 target_is_directory=(shulker_root / resource_path).is_dir(),
399 )
402def _rglob(root: Path, max_depth: int) -> Iterable[Path]:
403 """Find all files (and directories* and symlinks) in the path up to the
404 specified depth
406 Parameters
407 ----------
408 root : Path
409 The path to search
410 max_depth : int
411 The maximum number of levels to go
413 Returns
414 -------
415 list-like of paths
416 The files (and directories and symlinks) in the path up to that depth
418 Notes
419 -----
420 - Unlike an actual rglob, this method does not return any directories that
421 are not at the maximum depth
422 - Setting max_depth to 0 (or below) will return all files in the root, but
423 ***be warned*** that because this method follows symlinks, you can very
424 easily find yourself in an infinite loop
425 """
426 top_level = root.iterdir()
427 if max_depth == 1:
428 return top_level
429 return itertools.chain(
430 *(
431 _rglob(path, max_depth - 1) if path.is_dir() else (path,)
432 for path in top_level
433 )
434 )
437def cache_placements(
438 minecraft_root: Path, placements: dict[str, dict[Path, list[str]]]
439) -> None:
440 """Write placement record to file
442 Parameters
443 ----------
444 minecraft_root : Path
445 The root directory that your minecraft stuff (or, at least, the one
446 that's the parent of your EnderChest folder)
447 placements : dict
448 A record of placed links, as generated by `place_ender_chest`
449 """
450 cache_file = fs.place_cache(minecraft_root)
451 cache_file.write_text(
452 json.dumps(
453 {
454 instance_name: {
455 str(resource_path): shulker_boxes
456 for resource_path, shulker_boxes in instance_placements.items()
457 }
458 for instance_name, instance_placements in placements.items()
459 },
460 indent=4,
461 sort_keys=False,
462 )
463 )
464 PLACE_LOGGER.debug("Placement cache written to %s", cache_file)
467def load_placement_cache(minecraft_root: Path) -> dict[str, dict[Path, list[str]]]:
468 """Load the placement cache from file
470 Parameters
471 ----------
472 minecraft_root : Path
473 The root directory that your minecraft stuff (or, at least, the one
474 that's the parent of your EnderChest folder)
476 Returns
477 -------
478 dict
479 A record of the placed symlinks, structured as a nested dict, matching
480 the schema of one generated by `place_ender_chest`
482 Raises
483 ------
484 OSError
485 If the placement cache could not be found, read or parsed
486 """
487 try:
488 cache_file = fs.place_cache(minecraft_root)
489 INVENTORY_LOGGER.debug(
490 "Loading placement cache from %s", fs.place_cache(minecraft_root)
491 )
492 raw_dict: dict[str, dict[str, list[str]]] = json.loads(
493 cache_file.read_text("UTF-8")
494 )
495 except json.JSONDecodeError as decode_error:
496 raise OSError(
497 f"{fs.place_cache(minecraft_root)} is corrupted and could not be parsed:"
498 ) from decode_error
499 return {
500 instance_name: {
501 Path(resource_path): shulker_boxes
502 for resource_path, shulker_boxes in instance_placements.items()
503 }
504 for instance_name, instance_placements in raw_dict.items()
505 }
508def trace_resource(
509 minecraft_root: Path,
510 pattern: str,
511 placements: dict[str, dict[Path, list[str]]],
512 instance_name: str | None = None,
513) -> list[tuple[Path, Path, list[str]]]:
514 """Given a filename or glob pattern, return a list of all matching
515 EnderChest-placed symlinks, together with a trace-back of the shulker boxes
516 each link targets
518 Parameters
519 ----------
520 minecraft_root : Path
521 The root directory that your minecraft stuff (or, at least, the one
522 that's the parent of your EnderChest folder)
523 pattern : filename, path or glob pattern
524 The resource to trace
525 placements : dict
526 A record of placed symlinks, such as the one generated by `place_ender_chest`.
527 instance_name : str, optional
528 The name of the instance to search. This variable is case-sensitive.
529 If None is given, all instances will be searched.
531 Returns
532 -------
533 list of (Path, Path, list) tuples
534 - The first item in each list is the path of the instance root
535 - The second item in each list is the path to a linked resource
536 matching the provided pattern (and instance), relative to the instance
537 root
538 - The third item is the list of shulker boxes, sorted in ascending
539 priority, into which that symlink was linked (explicitly, the
540 _last_ entry in each list corresponds to the shulker box inside
541 which that link currently points)
543 Raises
544 ------
545 OSError
546 If no placement cache was provided and the placement cache file could
547 not be found, read or parsed
548 KeyError
549 If there is no instance registered to this EnderChest with the specified
550 name
551 """
552 instances = {
553 instance.name: instance
554 for instance in load_ender_chest_instances(
555 minecraft_root, log_level=logging.DEBUG
556 )
557 }
558 if instance_name is None:
559 return sum(
560 (
561 trace_resource(minecraft_root, pattern, placements, name)
562 for name in instances
563 ),
564 [],
565 )
566 instance_root = instances[instance_name].root
567 matches: list[tuple[Path, Path, list[str]]] = []
568 for resource_path, target_boxes in placements[instance_name].items():
569 if (
570 fnmatch.fnmatchcase(str(resource_path), pattern)
571 or fnmatch.fnmatchcase(str(resource_path), os.path.join("*", pattern))
572 or fnmatch.fnmatchcase(
573 os.path.abspath(minecraft_root / instance_root / resource_path),
574 os.path.join("*", pattern),
575 )
576 ):
577 matches.append((instance_root, resource_path, target_boxes))
578 return matches
581def report_resource_trace(
582 minecraft_root: Path, instance_root: Path, resource_path: Path, boxes: Sequence[str]
583) -> None:
584 """Print (log) the shulker boxes an instance resource is linked to
586 Parameters
587 ----------
588 minecraft_root : Path
589 The root directory that your minecraft stuff (or, at least, the one
590 that's the parent of your EnderChest folder)
591 instance_root : Path
592 The path of the EnderChest-placed symlink
593 resource_path : Path
594 The path to the symlink, relative to the instance root
595 boxes : list of str
596 The names of the shulker boxes, sorted by ascending priority, that are
597 targeted by this symlink (technically only the last entry in this list
598 is the actual target)
599 """
600 symlink_location = instance_root / resource_path
601 if len(boxes) == 0: # pragma: no cover
602 # Since defaultdicts are involved, this could happen accidentally at
603 # some point and should just be ignored
604 return
605 *other_box_names, primary_box_name = boxes
606 try:
607 INVENTORY_LOGGER.log(
608 IMPORTANT,
609 "%s currently resolves to %s",
610 symlink_location,
611 os.path.abspath(
612 (
613 symlink_location / (minecraft_root / symlink_location).readlink()
614 ).expanduser()
615 ),
616 )
617 except OSError:
618 INVENTORY_LOGGER.warning(
619 "%s no longer exists or is not a symlink", symlink_location
620 )
622 INVENTORY_LOGGER.log(
623 IMPORTANT,
624 " based on being linked into shulker box: %s",
625 primary_box_name,
626 )
627 INVENTORY_LOGGER.debug(
628 " - > %s",
629 fs.shulker_box_root(minecraft_root, primary_box_name) / resource_path,
630 )
632 for box_name in reversed(other_box_names):
633 INVENTORY_LOGGER.info(
634 " which overwrote the link into shulker box: %s", box_name
635 )
636 INVENTORY_LOGGER.debug(
637 " - > %s",
638 fs.shulker_box_root(minecraft_root, box_name) / resource_path,
639 )
642def list_placements(
643 minecraft_root: Path, pattern: str, instance_name: str | None = None
644) -> None:
645 """Report all shulker boxes that provide files matching the given pattern
647 Parameters
648 ----------
649 minecraft_root : Path
650 The root directory that your minecraft stuff (or, at least, the one
651 that's the parent of your EnderChest folder)
652 pattern : filename, path or glob pattern
653 The pattern of the resource to trace
654 instance_name : str, optional
655 The name of the instance to search. This variable is case-sensitive.
656 If None is given, all instances will be searched.
657 """
658 try:
659 placements = load_placement_cache(minecraft_root)
660 except OSError as no_cache:
661 INVENTORY_LOGGER.error(
662 "The placement cache could not be loaded:"
663 "\n %s"
664 "\nPlease run enderchest place again to regenerate the cache.",
665 no_cache,
666 )
667 return
668 try:
669 matches = trace_resource(
670 minecraft_root, pattern, placements, instance_name=instance_name
671 )
672 except KeyError:
673 INVENTORY_LOGGER.error(
674 "No instance named %s is registered to this EnderChest", instance_name
675 )
676 return
677 if len(matches) == 0:
678 INVENTORY_LOGGER.warning(
679 "Could not find any placed resources matching the pattern %s%s."
680 "\n\nNote: this command does not check inside linked folders.",
681 pattern,
682 f"\nin the instance {instance_name}" if instance_name else "",
683 )
684 return
685 for match in matches:
686 report_resource_trace(minecraft_root, *match)