index : sod

        

            
            
        
        
0#!/bin/python
1# MIT License
2#
3# Copyright (c) 2026 dev@ptrace.dev
4#
5# Permission is hereby granted, free of charge, to any person obtaining a copy
6# of this software and associated documentation files (the "Software"), to deal
7# in the Software without restriction, including without limitation the rights
8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9# copies of the Software, and to permit persons to whom the Software is
10# furnished to do so, subject to the following conditions:
11#
12# The above copyright notice and this permission notice shall be included in all
13# copies or substantial portions of the Software.
14#
15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21# SOFTWARE.
22#
23
24import pickle
25import sys
26import os
27import tarfile
28import shutil
29import subprocess
30import time
31from functools import wraps
32from enum import Enum
33from concurrent.futures import ThreadPoolExecutor
34from uuid import uuid4
35from argparse import ArgumentParser, RawDescriptionHelpFormatter
36from pathlib import Path
37from datetime import datetime
38
39
40
41PATH_HOME = Path.home()
42PATH_CONFIG = PATH_HOME / Path(".config")
43PATH_CONFIG_SOD_DIR = Path("sod-backup-helper")
44PATH_CONFIG_FILE = Path("sod.config")
45PATH_CONFIG_FILE_DOT = Path(".sod-config")
46
47
48SUCCESS = True
49FAILURE = False
50
51
52ARGP_DESCRIPTION = """
53Creates a backup from your supplied path and distributes it to specific directories.
54
55    Files getting archived and compressed using `tar` command.
56    Stored as `<parent dir name>_<year>_<month>_<day>_<uuid4>.tar.gz`
57
58    You can also change the exact pattern of the file name, like this:
59        sod -c "my_prefix_<year>_<uuid4>.tar.gz"
60
61    Note: It stores a config file in `~/.config/sod-backup-helper`
62          or `~/.sod-config`. It depends if the `.config` directory exists.
63
64"""
65
66HELP_PATH = ""
67HELP_P = "Adds backup paths. Example: `-p Foo Bar` or `-p Foo -p Bar`"
68HELP_L = "Lists all backup paths."
69HELP_R = "Removes the backup path by index. Index is supplied by the `-l` argument."
70HELP_N = "Removes ALL backup paths. Passing `-nn` removes the config file."
71HELP_C = "Define a new pattern. You can also use the following palceholders: <year> <month> <day> <uuid4>"
72HELP_Z = "Enable compression."
73HELP_X = "Don't create the archive, just print debug info."
74HELP_D = "Prints the current state."
75
76
77class Archive_Backend(Enum):
78    NONE    = 0
79    PYTHON  = 1
80    GNU     = 2
81
82
83class State:
84    def __init__(self):
85        self.backup_destinations = []
86        self.name_pattern = ""
87
88
89default_state = State()
90default_state.backup_destinations = []
91default_state.name_pattern = "<name>_%Y_%m_%d_<uuid>"
92
93
94class Args:
95    def __init__(self):
96        self.parser = ArgumentParser(
97            prog = "Safe our Data",
98            formatter_class = RawDescriptionHelpFormatter,
99            description = ARGP_DESCRIPTION
100        )
101        self.parser.add_argument("path", type=Path, nargs="?", default=None, help=HELP_PATH)
102        self.parser.add_argument("-p", "--backup-path-add",
103                                 action="extend", nargs="+", type=str, help=HELP_P)
104        self.parser.add_argument("-l", "--backup-path-list", action="store_true", help=HELP_L)
105        self.parser.add_argument("-r", "--backup-path-remove", type=int, help=HELP_R)
106        self.parser.add_argument("-n", "--backup-path-nuke", action="count", default=0, help=HELP_N)
107        self.parser.add_argument("-c", "--pattern-change", type=str, help=HELP_C)
108        self.parser.add_argument("-z", "--archive-compress", action="store_true", help=HELP_Z)
109        self.parser.add_argument("-x", "--debug-dry-run", action="store_true", help=HELP_X)
110        self.parser.add_argument("-d", "--debug", action="store_true", help=HELP_D)
111
112    def get(self):
113        return self.parser.parse_args()
114
115    def help(self):
116        self.parser.print_help()
117        return self
118
119
120args_handle = Args()
121args = args_handle.get()
122
123
124def timer(func):
125    @wraps(func)
126    def wrapper(*args, **kwargs):
127        start = time.perf_counter()
128        result = func(*args, **kwargs)
129        print(f"{func.__name__} took {time.perf_counter() - start:.6f}s")
130        return result
131    return wrapper
132
133
134def run():
135    config_exists, config_fp = config_find()
136
137    if not config_exists:
138        if not config_create_file(config_fp): return FAILURE
139
140    success, state = config_read(config_fp)
141    if not success: return FAILURE
142
143    if args.path:
144        return backup_do(state, args.path)
145
146    operation = SUCCESS
147
148    if args.backup_path_list:
149        backup_path_list(state)
150    elif cringe(args.pattern_change):
151        backup_name_fmt_new(state, args.pattern_change)
152    elif args.debug:
153        debug_view(state, config_fp)
154    elif cringe(args.backup_path_add):
155        operation = backup_path_add(state, args.backup_path_add)
156    elif cringe(args.backup_path_remove):
157        operation = backup_path_remove(state, args.backup_path_remove)
158    elif args.backup_path_nuke == 1:
159        backup_path_nuke(state)
160    elif args.backup_path_nuke == 2:
161        return config_delete(config_fp)
162
163    if operation == SUCCESS:
164        config_write(state, config_fp)
165
166    return operation
167
168
169def cringe(anything) -> bool:
170    return anything is not None
171
172
173def path_process_and_validate(state: State, path: Path) -> tuple[bool, Path, Path]:
174    if not path.exists():
175        print("This path does not exist:", path)
176        return FAILURE, Path(), Path()
177
178    stem = path.stem
179    archive_name = backup_name_create(state, stem)
180
181    return SUCCESS, archive_name, path
182
183
184@timer
185def archive_make(archive_fp, file_to_archive) -> bool:
186    backend = archive_backend_determine()
187
188    if args.debug_dry_run:
189        dbg("-- Archiving -------------")
190        dbg(f"archive_fp: {archive_fp}")
191        dbg(f"file_to_archive: {file_to_archive}")
192        dbg(f"backend: {backend}")
193        dbg(f"compression: {args.archive_compress}")
194        return SUCCESS
195
196    try:
197        if backend == Archive_Backend.GNU:
198            archive_backend_gnu(archive_fp, file_to_archive)
199        elif backend == Archive_Backend.PYTHON:
200            archive_backend_python(archive_fp, file_to_archive)
201        else:
202            print("This should not happen! Blame the developer!")
203            return FAILURE
204        return SUCCESS
205    except Exception as e:
206        print("Could not archive file:", e)
207        return FAILURE
208
209
210def archive_backend_python(archive_fp, file_to_archive):
211    compression = "w:gz" if args.archive_compress else "w:"
212    with tarfile.open(archive_fp, compression) as tar:
213        tar.add(file_to_archive)
214
215
216def archive_backend_gnu(archive_fp, file_to_archive):
217    compression = "-czf" if args.archive_compress else "-cf"
218    cmd = ["tar", compression, archive_fp, file_to_archive]
219    proc = subprocess.run(cmd, capture_output=True)
220
221    # fugly check, I know
222    if proc.returncode != 0:
223        print(proc.stderr)
224        print(proc.stdout)
225
226    proc.check_returncode()
227
228
229def archive_backend_determine() -> Archive_Backend:
230    probe_tar_cmd = ["tar", "--version"]
231    s = subprocess.run(probe_tar_cmd, capture_output=True)
232
233    if s.returncode == 0:
234        return Archive_Backend.GNU
235    else:
236        return Archive_Backend.PYTHON
237
238
239def make_directory_if_not_exists(fp: Path) -> bool:
240    if args.debug_dry_run:
241        dbg(f"mkdir: {fp}")
242        return SUCCESS
243    try:
244        fp.mkdir(exist_ok=True)
245        return SUCCESS
246    except Exception as e:
247        print("Could not create directory:", e)
248        return FAILURE
249
250
251def backup_do(state: State, fp) -> bool:
252    if len(state.backup_destinations) == 0:
253        print("No backup path defined. Canceling ...")
254        return FAILURE
255
256    operation, archive_name, unarchived_file_path = path_process_and_validate(state, fp)
257    if not operation: return operation
258
259    first_backup_dir = state.backup_destinations[0]
260
261    intermediate_path = Path(unarchived_file_path.stem)
262    archive_dir = first_backup_dir / intermediate_path
263    archive_fp = archive_dir / archive_name
264
265    if not make_directory_if_not_exists(archive_dir): return FAILURE
266    if not archive_make(archive_fp, unarchived_file_path): return FAILURE
267    if not backup_distribute(state, archive_name, intermediate_path): return FAILURE
268
269    print("Backups archived and distributed.")
270    return SUCCESS
271
272
273@timer
274def backup_distribute(state: State, archive_name, intermediate_path) -> bool:
275
276    def copy_file(src, dst):
277        shutil.copy(src, dst)
278
279    def distribute(src, destinations, max_workers=5):
280        with ThreadPoolExecutor(max_workers=max_workers) as executor:
281            futures = [executor.submit(copy_file, src, dst) for dst in destinations]
282            for future in futures:
283                future.result()
284
285
286    source_file = state.backup_destinations[0] / intermediate_path / archive_name
287
288    if not args.debug_dry_run and not source_file.exists():
289        print("[  Error  ]: Sanity check failed, could not find:", source_file)
290        return FAILURE
291
292    valid_backup_paths = []
293
294    for i in range(1, len(state.backup_destinations)):
295        path = state.backup_destinations[i] / intermediate_path
296
297        if not make_directory_if_not_exists(path):
298            print("[  Warning  ]: Skipping current operation.")
299            continue
300
301        valid_backup_paths.append(path / archive_name)
302
303    if args.debug_dry_run:
304        dbg("-- Distribute ------------")
305        dbg(f"source_file: {source_file}")
306        dbg("valid_backup_paths:")
307        for dbg_paths in valid_backup_paths:
308            dbg(f"    {dbg_paths}")
309        return SUCCESS
310
311    try:
312        distribute(source_file, valid_backup_paths)
313    except Exception as e:
314        print("Could not copy backups to other destinations:", e)
315        return FAILURE
316
317    return SUCCESS
318
319
320
321def backup_name_create(state: State, fn) -> Path:
322    as_date = datetime.now().strftime(state.name_pattern)
323    out = (as_date.replace("<name>", fn)
324                  .replace("<uuid>", str(uuid4())))
325    file_ext = "tar.gz" if args.archive_compress else "tar"
326    return Path(f"{out}.{file_ext}")
327
328
329def backup_name_fmt_new(state: State, fmt):
330    state.name_pattern = fmt
331    print("Created new pattern:", state.name_pattern)
332
333
334def backup_path_nuke(state: State):
335    if len(state.backup_destinations) == 0:
336        print("No paths in list")
337        return
338
339    safe_phrase = "I'm damn sure!"
340    print("Are you sure to remove every path? " \
341         f"Type `{safe_phrase}` to confirm it, or nothing to cancel:")
342
343    user = input("> ")
344
345    if user == safe_phrase:
346        state.backup_destinations = []
347        print("Removed all paths.")
348        return
349
350    print("Operation canceled.")
351
352
353def backup_path_remove(state: State, path_index):
354    length = len(state.backup_destinations)
355
356    if length == 0:
357        print("No paths in list")
358        return SUCCESS
359
360    if path_index < 0 or path_index > length-1 :
361        print("Invalid index. Can't be lower than zero or higher than", length-1)
362        return FAILURE
363
364    removed = state.backup_destinations.pop(path_index)
365    print("Removed:", removed)
366    return SUCCESS
367
368
369def backup_path_add(state: State, paths):
370    if not paths:
371        print("No paths supplied")
372        return FAILURE
373
374    d = state.backup_destinations
375
376    for path in paths:
377        p = Path(path).resolve()
378        if not p.exists():
379            print("[  Warning  ]: Skipped this path, since it does not exist:", p)
380            continue
381        if not p in d: d.append(p)
382
383    print("\nCurrent backup paths:")
384    backup_path_list(state)
385
386    return SUCCESS
387
388
389def backup_path_list(state: State):
390    if not state.backup_destinations:
391        print("No backup paths defined yet")
392
393    for idx, path in enumerate(state.backup_destinations):
394        print(f"  {idx}:", path)
395
396    print()
397
398
399def config_find() -> tuple[bool, Path]:
400    path_config_at_home = PATH_HOME / PATH_CONFIG_FILE_DOT
401    path_config_at_dotconfig = PATH_CONFIG / PATH_CONFIG_SOD_DIR / PATH_CONFIG_FILE
402
403    if PATH_CONFIG.exists():
404        if path_config_at_dotconfig.exists():
405            return SUCCESS, path_config_at_dotconfig
406        else:
407            return FAILURE, path_config_at_dotconfig
408
409    if path_config_at_home:
410        return SUCCESS, path_config_at_home
411
412    return FAILURE, path_config_at_home
413
414
415def config_create_file(config_fp: Path) -> bool:
416    try:
417        config_fp.parent.mkdir(exist_ok=True)
418        success = config_write(default_state, config_fp)
419
420        if not success: return FAILURE
421
422        print(f"Created configuration file at `{config_fp}`")
423        return SUCCESS
424    except Exception as e:
425        print("Error: Could not create file:", e)
426        return FAILURE
427
428
429def config_delete(config_fp: Path) -> bool:
430    safe_phrase = "YEEES!"
431
432    print("Are you sure you want to delete the configuration file?")
433    print(f"Type `{safe_phrase}` to delete, type nothing to abort:")
434
435    user = input("> ")
436
437    if user == safe_phrase:
438        try:
439            os.remove(config_fp)
440            print("Deleted:", config_fp)
441            return SUCCESS
442        except Exception as e:
443            print("Could not delete file:", e)
444            return FAILURE
445
446    print("Operation canceled.")
447    return SUCCESS
448
449
450def config_write(state: State, config_fp) -> bool:
451    try:
452        with open(config_fp, 'wb') as f:
453            pickle.dump(state, f)
454        return SUCCESS
455    except Exception as e:
456        print("Error: Could not write config file:", e)
457        return FAILURE
458
459
460def config_read(config_fp) -> tuple[bool, State]:
461    try:
462        with open(config_fp, 'rb') as f:
463            return SUCCESS, pickle.load(f)
464    except Exception as e:
465        print("Error: Could not read config file:", e)
466        return FAILURE, State()
467
468
469def debug_view(state: State, config_fp):
470    print("Config Path:", config_fp)
471    debug_display_state_file("State File ", state)
472    debug_display_state_file("Default Config ", default_state)
473
474
475def debug_display_state_file(label, state: State):
476    print(f"-- Debug: {label:-<40}")
477    print()
478
479    for k, v in state.__dict__.items():
480        if isinstance(v, list):
481            print(f"{' '*4}{k}:")
482            for e in v: print(f"{' '*8}{e}")
483            print()
484        else:
485            print(f"{' '*4}{k}:  {v}")
486
487    print()
488
489
490def dbg(msg):
491    print("[  DEBUG   ]:", msg)
492
493
494
495if __name__ == "__main__":
496    if len(sys.argv) == 1:
497        args_handle.help()
498        sys.exit(0)
499
500    if run():
501        sys.exit(0)
502    else:
503        sys.exit(-1)
504