dropbox conflicted file resolver

dropbox_resolve_conflicts.py

#!/usr/bin/env python
"""
Script to de-conflict Dropbox conflicted files

Usage:

    dropbox-fix-conflicts [options] <file_path>

    Options:
        -h --help           Show this screen.
        -q --quiet          Don't print status messages to stdout
        -v --verbose        Print debug messages
        -d --dry-run        Don't modify any files

Raises:
    DropboxConflictError: Error on creating a conflict model
    Exception: Other errors

Returns:
    int: number of files resolved
"""

import re
import os
import dataclasses
import sys
from pathlib import Path
from optparse import OptionParser
import platform
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

DROPBOX_FOLDER = os.path.expanduser("~/Dropbox/")

DROPBOX_FOLDER_BACKUP = os.path.join(DROPBOX_FOLDER, ".backups")

# @TODO read ignores
IGNORE_DIRECTORIES = [".git", ".mypycache", "node_modules"]

_local_hostname = platform.node()

_match_string_build = f"(?P<filename>.*)(?P<conflict>\ \({re.escape(_local_hostname)}\\'s\ conflicted\ copy\ \d{{4}}-\d{{2}}-\d{{2}}\))(?P<extension>.*)?"

_match_conflict = re.compile(_match_string_build)

class DropboxConflictError(Exception):
    """ Dropbox conflict error """
    pass

@dataclasses.dataclass
class DropboxConflictedFile:
    original: Path
    conflict: Path


def conflicted_file_is_newer(original_file: Path, conflict_file: Path) -> bool:
    """ Returns if original file is older than conflict file """
    _, _, _, _, _, _, _, _, original_mtime, _ = original_file.stat()
    _, _, _, _, _, _, _, _, conflict_mtime, _ = conflict_file.stat()

    # @TODO print human times
    logging.debug(f"Original: {original_mtime} Conflict: {conflict_mtime}")

    return conflict_mtime > original_mtime


def find_conflicted_files(path: Path, skip_hidden_directories: bool = True) -> list[DropboxConflictedFile]:
    """ Find all conflict files """
    ignored_folders = IGNORE_DIRECTORIES

    conflicted_files: list[DropboxConflictedFile] = []

    for root, dirs, files in os.walk(str(path)):
        dirs[:] = [d for d in dirs if d not in ignored_folders]

        if skip_hidden_directories:
            dirs[:] = [d for d in dirs if not d.startswith(".")]

        for subject_file_name in files:
            subject_file_path = Path(root) / Path(subject_file_name)

            if not subject_file_path.is_file():
                logging.debug(f"Skipping {subject_file_path} as it is not a file")
                continue

            if subject_match := re.match(_match_conflict, str(subject_file_path)):
                logging.debug(f"Found conflict file: {subject_file_path}")
                original_file_path = Path(subject_match.group("filename") + subject_match.group("extension"))

                if not original_file_path.exists():
                    raise DropboxConflictError(f"Original file {original_file_path} does not exist")

                try:
                    conflict_model = DropboxConflictedFile(original_file_path, subject_file_path)
                except ValueError as error:
                    logging.error(f"Error creating DropboxConflictedFile: {error}")
                    continue

                conflicted_files.append(conflict_model)

    return conflicted_files


def resolve_conflicted_files(conflicted_files: list[DropboxConflictedFile], dry_run: bool = True) -> None:
    """ Takes a list of conflicted files and resolves them. Will backup the old file if it is overwritten"""
    for conflicted_file in conflicted_files:
        if not conflicted_file_is_newer(conflicted_file.original, conflicted_file.conflict):
            continue

        backup_path = DROPBOX_FOLDER_BACKUP / conflicted_file.original

        logging.info(f"Backing up {conflicted_file.original} to {backup_path}")
        logging.info(f"Resolved conflict: {conflicted_file.conflict} -> {conflicted_file.original}")

        if not dry_run:
            # conflicted_file.original.rename(conflicted_file.original)
            conflicted_file.conflict.rename(conflicted_file.original)



def normalize_file_path(path: str) -> Path:
    """ Normalizes the file path """
    return Path(os.path.abspath(path))

def ensure_backup_directory() -> None:
    """ Ensures the backup directory exists """
    if not os.path.exists(DROPBOX_FOLDER_BACKUP):
        os.makedirs(DROPBOX_FOLDER_BACKUP)

def get_optparser() -> OptionParser:
    """ Command line option parser """
    parser = OptionParser()
    parser.add_option("-q", "--quiet",
                  action="store_false", dest="quiet", default=False,
                  help="don't print status messages to stdout")
    parser.add_option("-v", "--verbose",
                  action="store_false", dest="verbose", default=True,
                  help="print debug messages")
    parser.add_option("-d", "--dry-run",
                  action="store_false", dest="dry_run", default=False,
                  help="don't modify any files")

    return parser

def run_dropbox_deconflict_main() -> int:
    logging.info(f"Running dropbox deconflict on {_local_hostname} ")

    if not Path(DROPBOX_FOLDER).is_dir():
        raise Exception(f"Dropbox folder {DROPBOX_FOLDER} does not exist. Dropbox not installed")

    parser = get_optparser()
    (options, args) = parser.parse_args()

    if not args:
        raise Exception("Require a file path")

    file_path = args.pop()

    file_path = normalize_file_path(file_path)

    if not file_path.exists() or not file_path.is_dir():
        raise Exception(f"{file_path} does not exist or is not a direcotry")

    if options.verbose:
        logging.getLogger().setLevel(logging.DEBUG)

    if options.quiet:
        logging.getLogger().setLevel(logging.ERROR)

    conflicted_files = find_conflicted_files(file_path)

    if not conflicted_files:
        logging.info("No conflicted files found")
        return 0

    resolve_conflicted_files(conflicted_files, dry_run=options.dry_run)

if __name__ == "__main__":
    try:
        run_dropbox_deconflict_main()
    # except Exception as e:
        # logging.error(e)
    except KeyboardInterrupt:
        sys.exit(1)