Coverage for enderchest/enderchest.py: 99%
183 statements
« prev ^ index » next coverage.py v7.7.1, created at 2025-03-28 20:32 +0000
1"""Specification and configuration of an EnderChest"""
3from collections.abc import Iterable
4from dataclasses import dataclass
5from pathlib import Path
6from socket import gethostname
7from typing import Any
8from urllib.parse import ParseResult, urlparse
10from . import config as cfg
11from . import filesystem as fs
12from . import instance as i
13from . import sync
14from .loggers import CRAFT_LOGGER, GATHER_LOGGER, INVENTORY_LOGGER
15from .sync import abspath_from_uri
17_DEFAULTS = (
18 ("offer_to_update_symlink_allowlist", True),
19 ("sync_confirm_wait", 5),
20 ("place_after_open", True),
21 ("do_not_sync", ("EnderChest/enderchest.cfg", "EnderChest/.*", ".DS_Store")),
22 (
23 "shulker_box_folders",
24 (
25 "config",
26 "mods",
27 "resourcepacks",
28 "saves",
29 "shaderpacks",
30 ),
31 ),
32 ("standard_link_folders", ()),
33 (
34 "global_link_folders",
35 (
36 "backups",
37 "cachedImages",
38 "crash-reports",
39 "logs",
40 "replay_recordings",
41 "screenshots",
42 "schematics",
43 "config/litematica", # still worth having in case max_depth>2
44 ".bobby",
45 ),
46 ),
47)
50@dataclass(init=False, repr=False, eq=False)
51class EnderChest:
52 """Configuration of an EnderChest
54 Parameters
55 ----------
56 uri : URI or Path
57 The "address" of this EnderChest, ideally as it can be accessed from other
58 EnderChest installations, including both the path to where
59 the EnderChest folder can be found (that is, the parent of the
60 EnderChest folder itself, aka the "minecraft_root"), its net location
61 including credentials, and the protocol that should be used to perform
62 the syncing. All that being said, if just a path is provided, the
63 constructor will try to figure out the rest.
64 name : str, optional
65 A unique name to give to this EnderChest installation. If None is
66 provided, this will be taken from the hostname of the supplied URI.
67 instances : list-like of InstanceSpec, optional
68 The list of instances to register with this EnderChest installation
69 remotes : list-like of URI, or (URI, str) tuples
70 A list of other installations that this EnderChest should be aware of
71 (for syncing purposes). When a (URI, str) tuple is provided, the
72 second value will be used as the name/alias of the remote.
74 Attributes
75 ----------
76 name : str
77 The unique name of this EnderChest installation. This is most commonly
78 the computer's hostname, but one can configure multiple EnderChests
79 to coexist on the same system (either for the sake of having a "cold"
80 backup or for multi-user systems).
81 uri : str
82 The complete URI of this instance
83 root : Path
84 The path to this EnderChest folder
85 instances : list-like of InstanceSpec
86 The instances registered with this EnderChest
87 remotes : list-like of (ParseResult, str) pairs
88 The other EnderChest installations this EnderChest is aware of, paired
89 with their aliases
90 offer_to_update_symlink_allowlist : bool
91 By default, EnderChest will offer to create or update `allowed_symlinks.txt`
92 on any 1.20+ instances that do not already blanket allow links into
93 EnderChest. **EnderChest will never modify that or any other Minecraft
94 file without your express consent.** If you would prefer to edit these
95 files yourself (or simply not symlink your world saves), change this
96 parameter to False.
97 sync_confirm_wait : bool or int
98 The default behavior when syncing EnderChests is to first perform a dry
99 run of every sync operation and then wait 5 seconds before proceeding with the
100 real sync. The idea is to give the user time to interrupt the sync if
101 the dry run looks wrong. This can be changed by either raising or lowering
102 the value of confirm, by disabling the dry-run-first behavior entirely
103 (`confirm=False`) or by requiring that the user explicitly confirms
104 the sync (`confirm=True`). This default behavior can also be overridden
105 when actually calling the sync commands.
106 place_after_open: bool
107 By default, EnderChest will follow up any `enderchest open` operation
108 with an `enderchest place` to refresh any changed symlinks. This
109 functionality can be disabled by setting this parameter to False.
110 do_not_sync : list of str
111 Glob patterns of files that should not be synced between EnderChest
112 installations. By default, this list comprises `EnderChest/enderchest.cfg`,
113 any top-level folders starting with a "." (like .git) and
114 `.DS_Store` (for all you mac gamers).
115 shulker_box_folders : list of str
116 The folders that will be created inside each new shulker box
117 standard_link_folders : list of str
118 The default set of "link folders" when crafting a new shulker box
119 global_link_folders : list of str
120 The "global" set of "link folders," offered as a suggestion when
121 crafting a new shulker box
122 """
124 name: str
125 _uri: ParseResult
126 _instances: list[i.InstanceSpec]
127 _remotes: dict[str, ParseResult]
128 offer_to_update_symlink_allowlist: bool
129 sync_confirm_wait: bool | int
130 place_after_open: bool
131 do_not_sync: list[str]
132 shulker_box_folders: list[str]
133 standard_link_folders: list[str]
134 global_link_folders: list[str]
136 def __init__(
137 self,
138 uri: str | ParseResult | Path,
139 name: str | None = None,
140 remotes: (
141 Iterable[str | ParseResult | tuple[str, str] | tuple[ParseResult, str]]
142 | None
143 ) = None,
144 instances: Iterable[i.InstanceSpec] | None = None,
145 ):
146 for setting, value in _DEFAULTS:
147 setattr(self, setting, list(value) if isinstance(value, tuple) else value)
149 try:
150 if isinstance(uri, ParseResult):
151 self._uri = uri
152 elif isinstance(uri, Path):
153 self._uri = urlparse(uri.absolute().as_uri())
154 else:
155 self._uri = urlparse(uri)
156 except AttributeError as parse_problem: # pragma: no cover
157 raise ValueError(f"{uri} is not a valid URI") from parse_problem
159 if not self._uri.netloc:
160 self._uri = self._uri._replace(netloc=sync.get_default_netloc())
161 if not self._uri.scheme:
162 self._uri = self._uri._replace(scheme=sync.DEFAULT_PROTOCOL)
164 self.name = name or self._uri.hostname or gethostname()
166 self._instances = []
167 self._remotes = {}
169 for instance in instances or ():
170 self.register_instance(instance)
172 for remote in remotes or ():
173 if isinstance(remote, (str, ParseResult)):
174 self.register_remote(remote)
175 else:
176 self.register_remote(*remote)
178 @property
179 def uri(self) -> str:
180 return self._uri.geturl()
182 def __repr__(self) -> str:
183 return f"EnderChest({self.uri, self.name})"
185 @property
186 def root(self) -> Path:
187 return fs.ender_chest_folder(abspath_from_uri(self._uri), check_exists=False)
189 @property
190 def instances(self) -> tuple[i.InstanceSpec, ...]:
191 return tuple(self._instances)
193 def register_instance(self, instance: i.InstanceSpec) -> i.InstanceSpec:
194 """Register a new Minecraft installation
196 Parameters
197 ----------
198 instance : InstanceSpec
199 The instance to register
201 Returns
202 -------
203 InstanceSpec
204 The spec of the instance as it was actually registered (in case the
205 name changed or somesuch)
207 Notes
208 -----
209 - If the instance's name is already assigned to a registered instance,
210 this method will choose a new one
211 - If this instance shares a path with an existing instance, it will
212 replace that instance
213 """
214 matching_instances: list[i.InstanceSpec] = []
215 for old_instance in self._instances:
216 if i.equals(abspath_from_uri(self._uri), instance, old_instance):
217 matching_instances.append(old_instance)
218 self._instances.remove(old_instance)
220 instance = i.merge(*matching_instances, instance)
222 name = instance.name
223 counter = 0
224 taken_names = {old_instance.name for old_instance in self._instances}
225 while True:
226 if name not in taken_names:
227 break
228 counter += 1
229 name = f"{instance.name}.{counter}"
231 GATHER_LOGGER.debug(f"Registering instance {name} at {instance.root}")
232 self._instances.append(instance._replace(name=name))
233 return self._instances[-1]
235 @property
236 def remotes(self) -> tuple[tuple[ParseResult, str], ...]:
237 return tuple((remote, alias) for alias, remote in self._remotes.items())
239 def register_remote(
240 self, remote: str | ParseResult, alias: str | None = None
241 ) -> None:
242 """Register a new remote EnderChest installation (or update an existing
243 registry)
245 Parameters
246 ----------
247 remote : URI
248 The URI of the remote
249 alias : str, optional
250 an alias to give to this remote. If None is provided, the URI's hostname
251 will be used.
253 Raises
254 ------
255 ValueError
256 If the provided remote is invalid
257 """
258 try:
259 remote = remote if isinstance(remote, ParseResult) else urlparse(remote)
260 alias = alias or remote.hostname
261 if not alias: # pragma: no cover
262 raise AttributeError(f"{remote.geturl()} has no hostname")
263 GATHER_LOGGER.debug("Registering remote %s (%s)", remote.geturl(), alias)
264 self._remotes[alias] = remote
265 except AttributeError as parse_problem: # pragma: no cover
266 raise ValueError(f"{remote} is not a valid URI") from parse_problem
268 @classmethod
269 def from_cfg(cls, config_file: Path) -> "EnderChest":
270 """Parse an EnderChest from its config file
272 Parameters
273 ----------
274 config_file : Path
275 The path to the config file
277 Returns
278 -------
279 EnderChest
280 The resulting EnderChest
282 Raises
283 ------
284 ValueError
285 If the config file at that location cannot be parsed
286 FileNotFoundError
287 If there is no config file at the specified location
288 """
289 INVENTORY_LOGGER.debug("Reading config file from %s", config_file)
290 config = cfg.read_cfg(config_file)
292 # All I'm gonna say is that Windows pathing is the worst
293 path = urlparse(config_file.absolute().parent.parent.as_uri()).path
295 instances: list[i.InstanceSpec] = []
296 remotes: list[str | tuple[str, str]] = []
298 requires_rewrite = False
300 scheme: str | None = None
301 netloc: str | None = None
302 name: str | None = None
303 sync_confirm_wait: str | None = None
304 place_after_open: bool | None = None
305 offer_to_update_symlink_allowlist: bool = True
306 do_not_sync: list[str] | None = None
307 folder_defaults: dict[str, list[str] | None] = {
308 "shulker_box_folders": None,
309 "standard_link_folders": None,
310 "global_link_folders": None,
311 }
313 for section in config.sections():
314 if section == "properties":
315 scheme = config[section].get("sync-protocol")
316 netloc = config[section].get("address")
317 name = config[section].get("name")
318 sync_confirm_wait = config[section].get("sync-confirmation-time")
319 place_after_open = config[section].getboolean("place-after-open")
320 offer_to_update_symlink_allowlist = config[section].getboolean(
321 "offer-to-update-symlink-allowlist", True
322 )
323 if "do-not-sync" in config[section].keys():
324 do_not_sync = cfg.parse_ini_list(
325 config[section]["do-not-sync"] or ""
326 )
327 for setting in folder_defaults.keys():
328 setting_key = setting.replace("_", "-")
329 if setting_key in config[section].keys():
330 folder_defaults[setting] = cfg.parse_ini_list(
331 config[section][setting_key] or ""
332 )
333 elif section == "remotes":
334 for remote in config[section].items():
335 if remote[1] is None:
336 raise ValueError("All remotes must have an alias specified")
337 remotes.append((remote[1], remote[0]))
338 else:
339 # TODO: flag requires_rewrite if instance was normalized
340 instances.append(i.InstanceSpec.from_cfg(config[section]))
342 scheme = scheme or sync.DEFAULT_PROTOCOL
343 netloc = netloc or sync.get_default_netloc()
344 uri = ParseResult(
345 scheme=scheme, netloc=netloc, path=path, params="", query="", fragment=""
346 )
348 ender_chest = EnderChest(uri, name, remotes, instances)
349 if sync_confirm_wait is not None:
350 match sync_confirm_wait.lower():
351 case "true" | "prompt" | "yes" | "confirm":
352 ender_chest.sync_confirm_wait = True
353 case "false" | "no" | "skip":
354 ender_chest.sync_confirm_wait = False
355 case _:
356 try:
357 ender_chest.sync_confirm_wait = int(sync_confirm_wait)
358 except ValueError as bad_input:
359 raise ValueError(
360 "Invalid value for sync-confirmation-time:"
361 f" {sync_confirm_wait}"
362 ) from bad_input
363 if place_after_open is None:
364 INVENTORY_LOGGER.warning(
365 "This EnderChest does not have a value set for place-after-open."
366 "\nIt is being set to False for now. To enable this functionality,"
367 "\nedit the value in %s",
368 config_file,
369 )
370 ender_chest.place_after_open = False
371 requires_rewrite = True
372 else:
373 ender_chest.place_after_open = place_after_open
375 ender_chest.offer_to_update_symlink_allowlist = (
376 offer_to_update_symlink_allowlist
377 )
379 if do_not_sync is not None:
380 ender_chest.do_not_sync = do_not_sync
381 chest_cfg_exclusion = "/".join(
382 (fs.ENDER_CHEST_FOLDER_NAME, fs.ENDER_CHEST_CONFIG_NAME)
383 )
384 if chest_cfg_exclusion not in do_not_sync:
385 INVENTORY_LOGGER.warning(
386 "This EnderChest was not configured to exclude the EnderChest"
387 " config file from sync operations."
388 "\nThat is being fixed now."
389 )
390 ender_chest.do_not_sync.insert(0, chest_cfg_exclusion)
391 requires_rewrite = True
392 for setting in folder_defaults.keys():
393 if folder_defaults[setting] is None:
394 folder_defaults[setting] = dict(_DEFAULTS)[setting] # type: ignore
395 # requires_rewrite = True # though I'm considering it
396 setattr(ender_chest, setting, folder_defaults[setting])
398 if requires_rewrite:
399 ender_chest.write_to_cfg(config_file)
400 return cls.from_cfg(config_file)
401 return ender_chest
403 def write_to_cfg(self, config_file: Path | None = None) -> str:
404 """Write this EnderChest's configuration to INI
406 Parameters
407 ----------
408 config_file : Path, optional
409 The path to the config file, assuming you'd like to write the
410 contents to file
412 Returns
413 -------
414 str
415 An INI-syntax rendering of this EnderChest's config
417 Notes
418 -----
419 The "root" attribute is ignored for this method
420 """
421 properties: dict[str, Any] = {
422 "name": self.name,
423 "address": self._uri.netloc,
424 "sync-protocol": self._uri.scheme,
425 }
426 if self.sync_confirm_wait is True:
427 properties["sync-confirmation-time"] = "prompt"
428 else:
429 properties["sync-confirmation-time"] = self.sync_confirm_wait
431 for setting, _ in _DEFAULTS:
432 if setting == "sync_confirm_wait":
433 continue # already did this one
434 setting_key = setting.replace("_", "-")
435 properties[setting_key] = getattr(self, setting)
437 remotes: dict[str, str] = {name: uri.geturl() for uri, name in self.remotes}
439 instances: dict[str, dict[str, Any]] = {}
441 for instance in self.instances:
442 instances[instance.name] = {
443 "root": instance.root,
444 "minecraft-version": instance.minecraft_versions,
445 "modloader": instance.modloader,
446 "groups": instance.groups_,
447 "tags": instance.tags_,
448 }
450 config = cfg.dumps(
451 fs.ENDER_CHEST_CONFIG_NAME, properties, remotes=remotes, **instances
452 )
454 if config_file:
455 CRAFT_LOGGER.debug("Writing configuration file to %s", config_file)
456 config_file.write_text(config)
457 return config
460def create_ender_chest(minecraft_root: Path, ender_chest: EnderChest) -> None:
461 """Create an EnderChest based on the provided configuration
463 Parameters
464 ----------
465 minecraft_root : Path
466 The root directory that your minecraft stuff is in (or, at least, the
467 one inside which you want to create your EnderChest)
468 ender_chest : EnderChest
469 The spec of the chest to create
471 Notes
472 -----
473 - The "root" attribute of the EnderChest config will be ignored--instead
474 the EnderChest will be created at <minecraft_root>/EnderChest
475 - This method does not check to see if there is already an EnderChest set
476 up at the specified location--if one exists, its config will
477 be overwritten
478 """
479 root = fs.ender_chest_folder(minecraft_root, check_exists=False)
480 root.mkdir(exist_ok=True)
482 config_path = fs.ender_chest_config(minecraft_root, check_exists=False)
483 ender_chest.write_to_cfg(config_path)
484 CRAFT_LOGGER.info(f"EnderChest configuration written to {config_path}")