コマンドライン ファイル同期ツール

#!/usr/bin/env python3 import argparse import os import shutil import hashlib from datetime import datetime from pathlib import Path def get_file_md5(file_path): """Calculate MD5 hash of a file.""" hash_md5 = hashlib.md5() try: with open(file_path, 'rb') as f: for chunk in iter(lambda: f.read(4096), b''): hash_md5.update(chunk) return hash_md5.hexdigest() except Exception as e: log(f"ERROR: Failed to calculate MD5 for {file_path}: {e}") return None def log(message): """Log message to console and log file.""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_message = f"[{timestamp}] {message}" print(log_message) if log_file_path: try: with open(log_file_path, 'a') as f: f.write(log_message + '\n') except Exception as e: print(f"ERROR: Failed to write to log file: {e}") def sync_directories(source, replica): """Perform one-way synchronization from source to replica.""" source_path = Path(source) replica_path = Path(replica) if not source_path.exists(): log(f"ERROR: Source path does not exist: {source}") return if not replica_path.exists(): log(f"Creating replica directory: {replica}") replica_path.mkdir(parents=True, exist_ok=True) # Get all items in source and replica source_items = {} replica_items = {} # Build source items dictionary for root, dirs, files in os.walk(source_path): for file in files: file_path = Path(root) / file rel_path = file_path.relative_to(source_path) source_items[str(rel_path)] = ('file', file_path) for dir_name in dirs: dir_path = Path(root) / dir_name rel_path = dir_path.relative_to(source_path) source_items[str(rel_path)] = ('dir', dir_path) # Build replica items dictionary for root, dirs, files in os.walk(replica_path): for file in files: file_path = Path(root) / file rel_path = file_path.relative_to(replica_path) replica_items[str(rel_path)] = ('file', file_path) for dir_name in dirs: dir_path = Path(root) / dir_name rel_path = dir_path.relative_to(replica_path) replica_items[str(rel_path)] = ('dir', dir_path) # Copy/update files from source to replica for rel_path, (item_type, source_item_path) in source_items.items(): replica_item_path = replica_path / rel_path if item_type == 'file': if rel_path not in replica_items: # File doesn't exist in replica, copy it replica_item_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source_item_path, replica_item_path) log(f"COPY {rel_path}") else: # File exists in replica, check if content is different source_md5 = get_file_md5(source_item_path) replica_md5 = get_file_md5(replica_item_path) if source_md5 and replica_md5 and source_md5 != replica_md5: shutil.copy2(source_item_path, replica_item_path) log(f"UPDATE {rel_path}") elif item_type == 'dir': if rel_path not in replica_items: # Directory doesn't exist in replica, create it replica_item_path.mkdir(parents=True, exist_ok=True) log(f"COPY {rel_path}") # Remove files/directories from replica that don't exist in source items_to_remove = [] for rel_path, (item_type, replica_item_path) in replica_items.items(): if rel_path not in source_items: items_to_remove.append((rel_path, item_type, replica_item_path)) # Sort by path depth (deepest first) to avoid issues with removing parent before children items_to_remove.sort(key=lambda x: x[0].count(os.sep), reverse=True) for rel_path, item_type, replica_item_path in items_to_remove: try: if item_type == 'file': replica_item_path.unlink() log(f"REMOVE {rel_path}") elif item_type == 'dir': shutil.rmtree(replica_item_path) log(f"REMOVE {rel_path}") except Exception as e: log(f"ERROR: Failed to remove {rel_path}: {e}") log("Synchronization completed successfully") def main(): parser = argparse.ArgumentParser( description='One-way file synchronization tool' ) parser.add_argument( 'source_path', help='Path to the source directory' ) parser.add_argument( 'replica_path', help='Path to the replica directory' ) parser.add_argument( 'log_file_path', help='Path to the log file' ) args = parser.parse_args() global log_file_path log_file_path = args.log_file_path # Clear or create log file try: with open(log_file_path, 'w') as f: f.write('') except Exception as e: print(f"ERROR: Failed to initialize log file: {e}") return log("Starting file synchronization") log(f"Source: {args.source_path}") log(f"Replica: {args.replica_path}") sync_directories(args.source_path, args.replica_path) if __name__ == '__main__': main()

import argparse import hashlib import logging import os import shutil import sys from pathlib import Path def setup_logger(log_file_path: Path) -> logging.Logger: logger = logging.getLogger("file_sync") logger.setLevel(logging.INFO) logger.handlers.clear() logger.propagate = False formatter = logging.Formatter("%(asctime)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S") console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) file_handler = logging.FileHandler(log_file_path, encoding="utf-8") file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger def md5sum(file_path: Path, chunk_size: int = 1024 * 1024) -> str: hasher = hashlib.md5() with file_path.open("rb") as f: while True: chunk = f.read(chunk_size) if not chunk: break hasher.update(chunk) return hasher.hexdigest() def files_are_different(source_file: Path, replica_file: Path) -> bool: if not replica_file.exists() or not replica_file.is_file(): return True if source_file.stat().st_size != replica_file.stat().st_size: return True return md5sum(source_file) != md5sum(replica_file) def copy_file(source_file: Path, replica_file: Path, logger: logging.Logger, operation: str) -> None: replica_file.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source_file, replica_file) logger.info(f"{operation} {replica_file}") def remove_path(path: Path, logger: logging.Logger) -> None: if path.is_dir() and not path.is_symlink(): shutil.rmtree(path) else: path.unlink() logger.info(f"REMOVE {path}") def sync_directories(source_root: Path, replica_root: Path, logger: logging.Logger) -> None: replica_root.mkdir(parents=True, exist_ok=True) source_entries = {p.relative_to(source_root) for p in source_root.rglob("*")} replica_entries = {p.relative_to(replica_root) for p in replica_root.rglob("*")} for rel_path in sorted(source_entries, key=lambda p: (len(p.parts), str(p))): source_path = source_root / rel_path replica_path = replica_root / rel_path if source_path.is_dir(): if replica_path.exists() and not replica_path.is_dir(): remove_path(replica_path, logger) if not replica_path.exists(): replica_path.mkdir(parents=True, exist_ok=True) logger.info(f"COPY {replica_path}") elif source_path.is_file(): if replica_path.exists() and replica_path.is_dir(): remove_path(replica_path, logger) if not replica_path.exists(): copy_file(source_path, replica_path, logger, "COPY") elif files_are_different(source_path, replica_path): copy_file(source_path, replica_path, logger, "UPDATE") for rel_path in sorted(replica_entries - source_entries, key=lambda p: (-len(p.parts), str(p))): replica_path = replica_root / rel_path if replica_path.exists(): remove_path(replica_path, logger) def validate_paths(source_path: Path, replica_path: Path, log_file_path: Path) -> None: if not source_path.exists() or not source_path.is_dir(): raise ValueError(f"Source path must exist and be a directory: {source_path}") if source_path.resolve() == replica_path.resolve(): raise ValueError("Source path and replica path must be different.") try: replica_resolved = replica_path.resolve(strict=False) source_resolved = source_path.resolve() if replica_resolved == source_resolved or source_resolved in replica_resolved.parents: raise ValueError("Replica path must not be the same as or inside the source path.") except Exception: pass if log_file_path.parent and not log_file_path.parent.exists(): log_file_path.parent.mkdir(parents=True, exist_ok=True) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="One-way file synchronization tool.") parser.add_argument("source_path", help="Path to the source directory") parser.add_argument("replica_path", help="Path to the replica directory") parser.add_argument("log_file_path", help="Path to the log file") return parser.parse_args() def main() -> int: args = parse_args() source_path = Path(args.source_path).expanduser() replica_path = Path(args.replica_path).expanduser() log_file_path = Path(args.log_file_path).expanduser() try: validate_paths(source_path, replica_path, log_file_path) logger = setup_logger(log_file_path) sync_directories(source_path, replica_path, logger) return 0 except Exception as exc: logger = logging.getLogger("file_sync_error") if not logger.handlers: logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", stream=sys.stdout, ) logger = logging.getLogger("file_sync_error") logger.error(f"ERROR {exc}") return 1 if __name__ == "__main__": sys.exit(main())