Coverage for enderchest/remote.py: 86%
85 statements
« prev ^ index » next coverage.py v7.10.1, created at 2025-07-30 12:06 +0000
1"""Higher-level functionality around synchronizing with different EnderCherts"""
3import logging
4from collections.abc import Sequence
5from pathlib import Path
6from time import sleep
7from urllib.parse import ParseResult, urlparse
9from . import filesystem as fs
10from . import inventory, place
11from .enderchest import EnderChest
12from .loggers import IMPORTANT, SYNC_LOGGER
13from .prompt import confirm
14from .sync import abspath_from_uri, pull, push, remote_file, render_remote
17def load_remote_ender_chest(uri: str | ParseResult) -> EnderChest:
18 """Load an EnderChest configuration from another machine
20 Parameters
21 ----------
22 uri : URI
23 The URI to the remote Minecraft root
25 Returns
26 -------
27 EnderChest
28 The remote EnderChest configuration
30 Raises
31 ------
32 ValueError
33 If the provided URI is invalid
34 RuntimeError
35 If the config from the remote EnderChest could not be parsed
36 """
37 try:
38 uri = uri if isinstance(uri, ParseResult) else urlparse(uri)
39 except AttributeError as bad_uri:
40 raise ValueError(f"{uri} is not a valid URI") from bad_uri
42 remote_root = Path(uri.path)
43 remote_config_path = fs.ender_chest_config(remote_root, check_exists=False)
44 uri = uri._replace(path=remote_config_path.as_posix())
46 with remote_file(uri) as remote_config:
47 try:
48 return EnderChest.from_cfg(remote_config)
49 except ValueError as bad_chest:
50 raise RuntimeError(
51 "The remote EnderChest config downloaded"
52 f"from {uri.geturl()} could not be parsed."
53 ) from bad_chest
56def fetch_remotes_from_a_remote_ender_chest(
57 uri: str | ParseResult,
58) -> list[tuple[ParseResult, str]]:
59 """Grab the list of EnderChests registered with the specified remote EnderChest
61 Parameters
62 ----------
63 uri : URI
64 The URI to the remote Minecraft root
66 Returns
67 -------
68 list of (URI, str) tuples
69 The URIs of the remote EnderChests, paired with their aliases
71 Raises
72 ------
73 RuntimeError
74 If the remote list could not be pulled
75 """
76 remote_chest = load_remote_ender_chest(uri)
77 remotes: list[tuple[ParseResult, str]] = [
78 (urlparse(uri) if isinstance(uri, str) else uri, remote_chest.name)
79 ]
81 remotes.extend(remote_chest.remotes)
82 SYNC_LOGGER.info(
83 "Loaded the following remotes:\n %s",
84 "\n".join(f" - {render_remote(alias, uri)}" for uri, alias in remotes),
85 )
87 if len(set(alias for _, alias in remotes)) != len(remotes):
88 raise RuntimeError(
89 f"There are duplicates aliases in the list of remotes pulled from {uri}"
90 )
91 return remotes
94def sync_with_remotes(
95 minecraft_root: Path,
96 pull_or_push: str,
97 dry_run: bool = False,
98 sync_confirm_wait: bool | int | None = None,
99 **sync_kwargs,
100) -> None:
101 """Pull changes from or push changes to remote EnderChests
103 Parameters
104 ----------
105 minecraft_root : Path
106 The root directory that your minecraft stuff (or, at least, the one
107 that's the parent of your EnderChest folder). This will be used to
108 construct relative paths.
109 pull_or_push : str
110 "pull" or "push"
111 dry_run: bool, optional
112 Perform a dry run of the sync operation, reporting the operations\
113 that will be performed but not actually carrying them out
114 sync_confirm_wait : bool or int, optional
115 The default behavior when syncing EnderChests is to first perform a dry
116 run of every sync operation and then wait 5 seconds before proceeding with the
117 real sync. The idea is to give the user time to interrupt the sync if
118 the dry run looks wrong. This can be changed by either raising or lowering
119 the value of confirm, by disabling the dry-run-first behavior entirely
120 (`confirm=False`) or by requiring that the user explicitly confirms
121 the sync (`confirm=True`). This default behavior can also be modified
122 in the EnderChest config. This parameter will be ignored when performing
123 a dry run.
124 sync_kwargs
125 Any additional arguments that should be passed into the syncing
126 operation
128 Notes
129 -----
130 - When pulling changes, this method will try each remote in the order they
131 are configured and stop once it has successfully pulled from a remote.
133 This method will attempt to push local changes to *every* remote
134 """
135 if pull_or_push not in ("pull", "push"):
136 raise ValueError(
137 'Invalid choice for sync operation. Choices are "pull" and "push"'
138 )
139 try:
140 if sync_confirm_wait is None:
141 sync_confirm_wait = inventory.load_ender_chest(
142 minecraft_root
143 ).sync_confirm_wait
144 this_chest = inventory.load_ender_chest(minecraft_root)
146 # I know this is redundant, but we want those logs
147 remotes = inventory.load_ender_chest_remotes(
148 minecraft_root, log_level=logging.DEBUG
149 )
150 except (FileNotFoundError, ValueError) as bad_chest:
151 SYNC_LOGGER.error(
152 "Could not load EnderChest from %s:\n %s", minecraft_root, bad_chest
153 )
154 return
155 if not remotes:
156 SYNC_LOGGER.error("EnderChest has no remotes. Aborting.")
157 return # kinda unnecessary
159 synced_somewhere = False
160 exclusions: Sequence[str] = sync_kwargs.pop("exclude", None) or ()
161 for remote_uri, alias in remotes:
162 if dry_run:
163 runs: tuple[bool, ...] = (True,)
164 elif sync_confirm_wait is False or sync_confirm_wait <= 0:
165 runs = (False,)
166 else:
167 runs = (True, False)
168 for do_dry_run in runs:
169 if dry_run:
170 prefix = "Simulating an attempt"
171 else:
172 prefix = "Attempting"
173 try:
174 remote_chest = load_remote_ender_chest(remote_uri)
175 if pull_or_push == "pull":
176 SYNC_LOGGER.log(
177 IMPORTANT,
178 "%s to pull changes from %s",
179 prefix,
180 render_remote(alias, remote_uri),
181 )
182 remote_chest_folder = remote_uri._replace(
183 path=urlparse(
184 (
185 fs.ender_chest_folder(
186 abspath_from_uri(remote_uri),
187 check_exists=False,
188 )
189 ).as_uri()
190 ).path
191 )
192 pull(
193 remote_chest_folder,
194 minecraft_root,
195 exclude={
196 *this_chest.do_not_sync,
197 *remote_chest.do_not_sync,
198 *exclusions,
199 },
200 dry_run=do_dry_run,
201 **sync_kwargs,
202 )
203 else:
204 SYNC_LOGGER.log(
205 IMPORTANT,
206 "%s to push changes to %s",
207 prefix,
208 render_remote(alias, remote_uri),
209 )
210 local_chest = fs.ender_chest_folder(minecraft_root)
211 push(
212 local_chest,
213 remote_uri,
214 exclude={
215 *this_chest.do_not_sync,
216 *remote_chest.do_not_sync,
217 *exclusions,
218 },
219 dry_run=do_dry_run,
220 **sync_kwargs,
221 )
222 except (
223 FileNotFoundError,
224 ValueError,
225 NotImplementedError,
226 TimeoutError,
227 RuntimeError,
228 ) as sync_fail:
229 SYNC_LOGGER.warning(
230 "Could not sync changes with %s:\n %s",
231 render_remote(alias, remote_uri),
232 sync_fail,
233 )
234 break
235 if do_dry_run == runs[-1]:
236 continue
237 if sync_confirm_wait is True:
238 if not confirm(default=True):
239 SYNC_LOGGER.error("Aborting")
240 return
241 else:
242 SYNC_LOGGER.debug("Waiting for %d seconds", sync_confirm_wait)
243 sleep(sync_confirm_wait)
244 else:
245 synced_somewhere = True
246 if pull_or_push == "pull":
247 if this_chest.place_after_open and not dry_run:
248 place.place_ender_chest(
249 minecraft_root,
250 keep_broken_links=False,
251 keep_stale_links=False,
252 error_handling="abort",
253 relative=False,
254 )
255 break
256 if not synced_somewhere:
257 SYNC_LOGGER.error("Could not sync with any remote EnderChests")