Index

s3-bsync / master

Bidirectional syncing tool to sync local filesystem directories with S3 buckets. (Incomplete)

Latest Commit

{#}TimeHashSubjectAuthor#(+)(-)GPG?
1528 Jun 2022 22:408506c93Update dataclass structuresJosh Stockin1108G

Blob @ s3-bsync / src / command_parse.py

application/x-python7343 bytesdownload raw
1# s3-bsync Copyright (c) 2021 Joshua Stockin
2# <https://joshstock.in>
3# <https://git.joshstock.in/s3-bsync>
4#
5# This software is licensed and distributed under the terms of the MIT License.
6# See the MIT License in the LICENSE file of this project's root folder.
7#
8# This comment block and its contents, including this disclaimer, MUST be
9# preserved in all copies or distributions of this software's source.
10
11import os
12import re
13import argparse
14import logging
15
16from .meta import package_info
17
18logger = logging.getLogger(__name__)
19
20__all__ = ["command_parse", "sanitize_arguments"]
21
22
23def command_parse(args: list[str]):
24 class Formatter(
25 argparse.RawDescriptionHelpFormatter,
26 argparse.ArgumentDefaultsHelpFormatter,
27 argparse.HelpFormatter,
28 ):
29 pass
30
31 parser = argparse.ArgumentParser(
32 prog=package_info["name"],
33 description=package_info["description"],
34 formatter_class=lambda prog: Formatter(prog, width=88),
35 allow_abbrev=False,
36 add_help=False,
37 )
38
39 parser.add_argument(
40 "--help", "-h", "-?", action="help", help="Display this help message and exit."
41 )
42 parser.add_argument(
43 "--version",
44 "-v",
45 action="version",
46 help="Display program and version information and exit.",
47 version=f"s3-bsync version {package_info['version_string']}\n"
48 "<https://joshstock.in> <josh@joshstock.in>\n"
49 "<https://git.joshstock.in/s3-bsync>",
50 )
51
52 group1 = parser.add_argument_group(
53 "program behavior", "The program runs in sync mode by default."
54 )
55
56 group1.add_argument(
57 "--init",
58 "-i",
59 action="store_true",
60 default=False,
61 help="Run in initialize (edit) mode. This allows tracking file management and directory options to be used.",
62 )
63 group1.add_argument(
64 "--debug",
65 action="store_true",
66 default=False,
67 help="Enables debug flag, which prints program information to stdout.",
68 )
69 group1.add_argument(
70 "--dryrun",
71 action="store_true",
72 default=False,
73 help="Run program logic without making changes. Useful when paired with debug mode to see what changes would be made.",
74 )
75
76 group2 = parser.add_argument_group(
77 "tracking file management", "Configuring the tracking file."
78 )
79
80 group2.add_argument(
81 "--file",
82 metavar=("SYNCFILE"),
83 default="~/.state.s3sync",
84 help="The s3sync state file used to store tracking and state information. It should resolve to an absolute path.",
85 )
86 group2.add_argument(
87 "--dump",
88 action="store_true",
89 default=False,
90 help="Dump s3sync state file configuration and exit.",
91 )
92 group2.add_argument(
93 "--purge",
94 action="store_true",
95 default=False,
96 help="Deletes the tracking configuration file if it exists and exits. Requires init mode.",
97 )
98 group2.add_argument(
99 "--overwrite",
100 action="store_true",
101 default=False,
102 help="Overwrite tracking file with new directory maps instead of appending. Requires init mode.",
103 )
104
105 group3 = parser.add_argument_group(
106 "directory mapping", "Requires initialize mode to be enabled."
107 )
108
109 group3.add_argument(
110 "--dir",
111 action="append",
112 nargs=2,
113 metavar=("PATH", "S3_DEST"),
114 default=argparse.SUPPRESS,
115 help="Directory map to detail which local directory corresponds to S3 bucket "
116 "and key prefix. Can be used multiple times to set multiple directories. "
117 "Local directories must be absolute. S3 destination in `s3://bucket-name/prefix` "
118 "format. Example: `--dir /home/josh/Documents s3://joshstockin/Documents`",
119 )
120 group3.add_argument(
121 "--rmdir",
122 action="append",
123 nargs=2,
124 metavar=("RMPATH", "S3_DEST"),
125 default=argparse.SUPPRESS,
126 help="Remove tracked directory map by local directory identifier. Running "
127 "`--rmdir /home/josh/Documents s3://joshstockin/Documents` would remove the"
128 "directory map from the s3sync file and stop tracking/syncing that directory.",
129 )
130 return parser.parse_args(args)
131
132
133class sync_settings:
134 def __init__(self, **kwargs):
135 self.__dict__.update(kwargs)
136
137
138def sanitize_arguments(args: argparse.Namespace):
139 settings = sync_settings()
140
141 if args.debug:
142 logger.debug("DEBUG flag set")
143 settings.debug = True
144
145 logger.debug(f'User supplied tracking file "{args.file}". Sanitizing...')
146 whitespace_pattern = re.compile(r"\s+")
147 args.file = os.path.expanduser(args.file)
148 args.file = re.sub(whitespace_pattern, "", args.file)
149 if not args.file:
150 logger.error("Inputted tracking file path string is empty")
151 exit(1)
152 if not os.path.isabs(args.file):
153 logger.error("Inputted tracking file path is not an absolute path")
154 exit(1)
155 if os.path.isdir(args.file):
156 logger.error("Inputted tracking file path resolves to an existing directory")
157 exit(1)
158 logger.debug(f'Tracking file set to "{args.file}"')
159 settings.syncfile = args.file
160
161 if args.init:
162 logger.debug("INIT mode set")
163 settings.mode = ["INIT"]
164 if args.overwrite:
165 if args.init:
166 logger.debug("OVERWRITE mode set")
167 settings.mode.append("OVERWRITE")
168 else:
169 logger.error("OVERWRITE mode requires INIT mode")
170 exit(1)
171 if args.purge:
172 if args.init:
173 logger.debug("PURGE mode set")
174 settings.mode.append("PURGE")
175 else:
176 logger.error("PURGE mode requires INIT mode")
177 exit(1)
178 if args.dump:
179 logger.debug("DUMP mode set")
180 settings.mode = ["DUMP"]
181
182 if not hasattr(settings, "mode"):
183 logger.debug("No mode set. Enabling SYNC mode implicitly")
184 settings.mode = ["SYNC"]
185 if args.dryrun:
186 logger.debug("DRYRUN flag enabled")
187 settings.mode.append("DRYRUN")
188
189 if hasattr(args, "dir"):
190 if not args.init:
191 logger.error("--dir requires INIT mode")
192 exit(1)
193 settings.dirmaps = {}
194 for dirmap in args.dir:
195 if not os.path.isabs(dirmap[0]):
196 logger.error(f'Local path "{dirmap[0]}" is not absolute')
197 exit(1)
198 if not os.path.isdir(dirmap[0]):
199 logger.error(
200 f'Local path "{dirmap[0]}" does not resolve to an existing directory'
201 )
202 exit(1)
203 settings.dirmaps[os.path.realpath(dirmap[0])] = dirmap[1]
204
205 if hasattr(args, "rmdir"):
206 if not args.init:
207 logger.error("--rmdir requires INIT mode")
208 exit(1)
209 settings.rmdirs = {}
210 for rmdir in args.rmdir:
211 if not os.path.isabs(rmdir[0]):
212 logger.error(f'Local path "{rmdir[0]}" is not absolute')
213 exit(1)
214 if not os.path.isdir(rmdir[0]):
215 logger.error(
216 f'Local path "{rmdir[0]}" does not resolve to an existing directory'
217 )
218 exit(1)
219 settings.rmdirs[os.path.realpath(rmdir[0])] = rmdir[1]
220
221 return settings
222