A Python script to sync a local folder with an Internet Archive item.

sync_ia.py

#!/usr/bin/env python

import os
import sys
import math
from pathlib import Path
import argparse

import internetarchive as ia


def parse_args(args):
    parser = argparse.ArgumentParser(
        description=f"Syncs an Internet Archive item with a local directory.\n\n"
                    f"Default parameters are shown in [brackets].",
        formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument("-i", "--item_id", dest="item", type=str, required=True,
                        help=f"the identifier of the item to upload to"
                       )
    parser.add_argument("-d", "--item_dir", dest="item_dir", type=str, required=True,
                        help=f"the directory to sync to the item"
                       )
    parser.add_argument("-g", "--globs", dest="globs", type=str, required=False, default="*",
                        help=f"the glob pattens to match, seperated by \"|\" characters [\"*\"]"
                       )
    parser.add_argument("-n", "--new_item", dest="new_item", action="store_true",
                        help=f"make a new item instead of uploading to an existing one"
                       )
    parser.add_argument("-k", "--keep_directories", dest="keep_directories", action="store_true",
                        help=f"preserve the directory structure of the item directory"
                       )
    parser.add_argument("-s", "--smallest_first", dest="smallest_first", action="store_true",
                        help=f"upload the smallest files first"
                       )
    parser.add_argument("-f", "--force", dest="force", action="store_true",
                        help=f"force the upload even if there are outstanding tasks"
                       )
    parser.add_argument("-p", "--prompt_exit", dest="prompt_exit", action="store_true",
                        help=f"prompt the user to press Enter before exiting"
                       )

    parsed_args = parser.parse_args(args)
    
    # Interpret string arguments
    item_dir = Path(parsed_args.item_dir).resolve()
    if not item_dir.is_dir():
        parser.error(f"the directory \"{parsed_args.item_dir}\" does not exist")
    parsed_args.item_dir = item_dir
    
    item = ia.get_item(parsed_args.item)
    if parsed_args.new_item:
        if item.exists:
            parser.error(f"the Internet Archive item \"{parsed_args.item}\" already exists, remove the --new_item flag to upload to it")
    else:
        if not item.exists:
            parser.error(f"the Internet Archive item \"{parsed_args.item}\" does not exist, use the --new_item flag to create it")
    parsed_args.item = item
    
    parsed_args.globs = parsed_args.globs.split("|")
    
    return parsed_args


def get_relative_path(target_path, base_path):
    target_path = Path(target_path).resolve()
    base_path = Path(base_path).resolve()
    
    if target_path.is_relative_to(base_path):
        return target_path.relative_to(base_path)
    else:
        return target_path


def get_file_list(directory, glob_patterns, by_size=False):
    if type(glob_patterns) != list:
        raise TypeError("\"glob_paterns\" must be a list of strings")
    
    directory = Path(directory)
    
    file_list = []
    
    for glob_pattern in glob_patterns:
        for item in list(directory.rglob(glob_pattern)):
            if item.is_file():
                file_list.append({
                    "absolute": str(item),
                    "relative": str(get_relative_path(item, directory))
                })
    
    if by_size:
        print("Smallest first option set, sorting files by size...")
        file_list.sort(key=lambda x: os.path.getsize(x["absolute"]))
    
    return file_list


def file_list_to_dict(files):
    file_dict = {}
    for file in files:
        file_dict[file["relative"]] = file["absolute"]
    
    return file_dict


def list_tasks(tasks):
    num_tasks = len(tasks)
    
    for i, task in enumerate(tasks[::-1]):
        print(f"Task {i+1}/{num_tasks}:")
        print(f"  Command:\t\t{task.cmd}")
        print(f"  Submit Time:\t\t{task.submittime}")
        print(f"  Submitted By:\t\t{task.submitter}")
        print(f"  Server:\t\t{task.server}")
        print(f"  Priority:\t\t{task.priority}")
        print(f"  Status:\t\t{task.status}")


def list_files(files):
    total_size_mib = 0.0
    
    for file in files:
        size_mib = Path(file["absolute"]).stat().st_size / (1024 ** 2)
        total_size_mib += size_mib
        
        size_mib = math.ceil(size_mib * 100) / 100
        print(f"- {file["relative"]} | {size_mib:.2f} MiB")
    
    total_size_mib = math.ceil(total_size_mib * 100) / 100
    print(f"Total upload size: {total_size_mib:.2f} MiB")


def prompt_exit(message="", print_only=False):
    prompt_text = "Press Enter to exit . . ."
    status_text = "RESULT:"
    
    if print_only:
        print(f"{status_text} {message}")
    else:
        if len(message) == 0:
            message = prompt_text
        else:
            message += f" {prompt_text}"
        
        input(message)
    
    sys.exit()


def ask_question(question):
    reply = input(f"{question} (y/n): ").strip().lower()
    
    if reply in ['y', 'yes']:
        return True
    else:
        return False


def main(args):
    print()
    parsed_args = parse_args(args)
    
    command_dir = Path.cwd()
    
    print(f"Item Identifier:\t{parsed_args.item.identifier}")
    print(f"Sync Directory:\t\t{get_relative_path(parsed_args.item_dir, command_dir)}")
    print(f"Glob Patterns:\t\t{" | ".join(parsed_args.globs)}")
    print()
    
    if parsed_args.force:
        print("Force option set, not checking for outstanding tasks...")
    else:
        print("Checking for outstanding tasks... ")
        tasks = parsed_args.item.get_catalog()
        if len(tasks) > 0:
            print("ERROR: Outstanding tasks exist for this item!")
            print("To avoid sync issues, please wait a while and try again later.")
            print()
            list_tasks(tasks)
            print()
            prompt_exit("Operation aborted for safety.", print_only=not parsed_args.prompt_exit)
        else:
            print("No outstanding tasks found for this item! Proceeding to sync...")
    print()
    
    print("Getting file list...")
    file_list = get_file_list(parsed_args.item_dir, parsed_args.globs, by_size=parsed_args.smallest_first)
    if parsed_args.keep_directories:
        print("Keep directories option set, will preserve the local directory structure...")
        files_in = file_list_to_dict(file_list)
    else:
        files_in = [x["absolute"] for x in file_list]
    print()
    
    print(f"The following files will be uploaded to the item:")
    list_files(file_list)
    print()
    
    if not ask_question("Do you want to proceed?"):
        print()
        prompt_exit("Operation canceled by user.", print_only=not parsed_args.prompt_exit)
    print()
    
    try:
        parsed_args.item.upload(
            files_in,
            verbose=True,
            retries=120,
            retries_sleep=30,
            checksum=True,
            verify=True,
            headers={"x-archive-keep-old-version": "0"}
        )
        
        print()
        prompt_exit("Operation complete!", print_only=not parsed_args.prompt_exit)
    except Exception as e:
        print()
        print(f"ERROR: Upload failed with the following error: {e}")
        print()
        prompt_exit("Operation ended due to an error.", print_only=not parsed_args.prompt_exit)

if __name__ == "__main__":
    main(sys.argv[1:])