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

#!/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())