Implement autocopy_tool: CLI data copy tool with rsync and local transfer support

Co-authored-by: daohu527 <10419854+daohu527@users.noreply.github.com>

Author: Copilot

autocopy_tool/__init__.py

@@ -0,0 +1,25 @@
+source:
+  type: network
+  base_path: /mnt/autodrive_data
+  host: 10.10.10.5
+  username: tester
+  # password: secret  # Not recommended; prefer SSH key authentication
+  # ssh_key: ~/.ssh/id_rsa
+  protocol: rsync
+
+targets:
+  - /data/test_today/
+
+rules:
+  log:
+    path: log/
+    filter: "{module}/{date}"
+  bag:
+    path: bag/
+    filter: "{date}"
+  map:
+    path: map/
+    filter: "{name}"
+  conf:
+    path: conf/
+    filter: "{name}"

autocopy_tool/config.yml

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""Autonomous Data Copy Tool – CLI entry point.
+
+Examples::
+
+    python main.py --type log --date 2025-11-04 --module perception
+    python main.py --type bag --date 2025-11-04
+    python main.py --type map --name shanghai_ring
+    python main.py --type conf --name default --local
+"""
+import argparse
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import yaml
+
+from autocopy_tool.modules.filters import build_source_path
+from autocopy_tool.modules.local_transfer import local_copy
+from autocopy_tool.modules.rsync_transfer import rsync_copy
+from autocopy_tool.utils.logger import get_logger
+from autocopy_tool.utils.time_utils import today, validate_date
+
+logger = get_logger(__name__)
+
+_DEFAULT_CONFIG = Path(__file__).parent / "config.yml"
+
+
+def load_config(path: str = str(_DEFAULT_CONFIG)) -> dict:
+    """Load and return the YAML configuration file.
+
+    Args:
+        path: Path to the YAML configuration file.
+
+    Returns:
+        Parsed configuration as a dictionary.
+    """
+    with open(path, encoding="utf-8") as fh:
+        return yaml.safe_load(fh)
+
+
+def parse_args(argv=None) -> argparse.Namespace:
+    """Parse and return command-line arguments."""
+    parser = argparse.ArgumentParser(
+        description="Autonomous Data Copy Tool",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog=__doc__,
+    )
+    parser.add_argument(
+        "--type",
+        required=True,
+        choices=["log", "bag", "map", "conf"],
+        help="Data type to copy.",
+    )
+    parser.add_argument(
+        "--date",
+        default=None,
+        help="Date filter in YYYY-MM-DD format (default: today). Used for log/bag types.",
+    )
+    parser.add_argument(
+        "--module",
+        default=None,
+        help="Module name for log copies (e.g. perception).",
+    )
+    parser.add_argument(
+        "--name",
+        default=None,
+        help="Name identifier for map/conf copies.",
+    )
+    parser.add_argument(
+        "--target",
+        default=None,
+        help="Override destination directory (default: first target in config.yml).",
+    )
+    parser.add_argument(
+        "--config",
+        default=str(_DEFAULT_CONFIG),
+        help="Path to configuration YAML file.",
+    )
+    parser.add_argument(
+        "--local",
+        action="store_true",
+        help="Use local filesystem copy instead of rsync (source must be locally mounted).",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv=None) -> int:
+    """Main entry point.
+
+    Returns:
+        Exit code (0 on success, non-zero on failure).
+    """
+    args = parse_args(argv)
+
+    try:
+        cfg = load_config(args.config)
+    except FileNotFoundError:
+        logger.error("Configuration file not found: %s", args.config)
+        return 1
+    except yaml.YAMLError as exc:
+        logger.error("Failed to parse configuration file: %s", exc)
+        return 1
+
+    # Resolve date (default to today for date-based types)
+    date = args.date
+    if date is not None:
+        try:
+            date = validate_date(date)
+        except ValueError as exc:
+            logger.error("%s", exc)
+            return 1
+    else:
+        date = today()
+
+    # Build source path
+    src = build_source_path(
+        cfg,
+        args.type,
+        date=date,
+        module=args.module or "",
+        name=args.name or "",
+    )
+
+    # Determine destination
+    dst = args.target
+    if dst is None:
+        targets = cfg.get("targets", [])
+        if not targets:
+            logger.error("No targets defined in configuration.")
+            return 1
+        dst = targets[0]
+
+    source_cfg = cfg["source"]
+
+    try:
+        if args.local:
+            local_copy(src, dst)
+        else:
+            rsync_copy(
+                src=src,
+                dst=dst,
+                host=source_cfg["host"],
+                user=source_cfg["username"],
+                ssh_key=source_cfg.get("ssh_key"),
+            )
+    except (FileNotFoundError, PermissionError, subprocess.CalledProcessError, OSError) as exc:
+        logger.error("Transfer failed: %s", exc)
+        return 1
+
+    logger.info("✅ Copy completed: %s -> %s", src, dst)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

autocopy_tool/main.py

@@ -0,0 +1,37 @@
+"""Path filter helpers for autocopy_tool.
+
+Builds source sub-paths from rule patterns defined in ``config.yml``.
+"""
+import os
+from typing import Any
+
+
+def build_source_path(cfg: dict, data_type: str, **kwargs: Any) -> str:
+    """Build the full source path for a given data type and filter parameters.
+
+    The rule's ``filter`` field is a Python format string whose placeholders
+    are populated from *kwargs*.  Unknown or missing keys default to an empty
+    string so that optional parameters (e.g. ``module`` for ``bag`` transfers)
+    do not raise a :class:`KeyError`.
+
+    Args:
+        cfg: Parsed configuration dictionary.
+        data_type: One of ``log``, ``bag``, ``map``, ``conf``.
+        **kwargs: Filter parameters (``date``, ``module``, ``name``, …).
+
+    Returns:
+        Absolute source path string.
+
+    Raises:
+        KeyError: If *data_type* is not present in ``cfg["rules"]``.
+    """
+    rule = cfg["rules"][data_type]
+    # Fill template; missing keys default to empty string
+    class _DefaultDict(dict):
+        def __missing__(self, key):
+            return ""
+
+    pattern = rule["filter"].format_map(_DefaultDict(kwargs))
+    # Strip any leading slashes that result from empty placeholder values
+    # so that os.path.join does not treat the pattern as an absolute path.
+    return os.path.join(cfg["source"]["base_path"], rule["path"], pattern.lstrip("/"))

autocopy_tool/modules/__init__.py

@@ -0,0 +1,40 @@
+"""Local file copy transfer module for autocopy_tool.
+
+Used when the source is already accessible on the local filesystem
+(e.g. a mounted NAS share or a USB drive).
+"""
+import os
+import shutil
+from pathlib import Path
+
+from autocopy_tool.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+def local_copy(src: str, dst: str) -> None:
+    """Copy files or a directory tree from a local source to a destination.
+
+    If *src* is a directory the entire tree is copied recursively.
+    If *src* is a file it is copied into *dst* (creating *dst* if needed).
+
+    Args:
+        src: Source file or directory path.
+        dst: Destination directory path.
+
+    Raises:
+        FileNotFoundError: If *src* does not exist.
+    """
+    src_path = Path(src)
+    if not src_path.exists():
+        raise FileNotFoundError(f"Source path does not exist: {src}")
+
+    os.makedirs(dst, exist_ok=True)
+
+    if src_path.is_dir():
+        dest_path = Path(dst) / src_path.name
+        shutil.copytree(src, dest_path, dirs_exist_ok=True)
+        logger.info("Directory copied: %s -> %s", src, dest_path)
+    else:
+        shutil.copy2(src, dst)
+        logger.info("File copied: %s -> %s", src, dst)

autocopy_tool/modules/filters.py

@@ -0,0 +1,50 @@
+"""rsync-based file transfer module for autocopy_tool."""
+import os
+import shlex
+import subprocess
+from typing import Optional
+
+from autocopy_tool.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+def rsync_copy(
+    src: str,
+    dst: str,
+    host: str,
+    user: str,
+    ssh_key: Optional[str] = None,
+    extra_args: Optional[list] = None,
+) -> None:
+    """Transfer files from a remote host using rsync over SSH.
+
+    Args:
+        src: Source path on the remote host.
+        dst: Destination directory on the local host.
+        host: Remote hostname or IP address.
+        user: Remote username.
+        ssh_key: Optional path to an SSH private key file.
+        extra_args: Optional list of additional rsync arguments.
+
+    Raises:
+        subprocess.CalledProcessError: If rsync exits with a non-zero status.
+    """
+    os.makedirs(dst, exist_ok=True)
+
+    ssh_parts = ["ssh"]
+    if ssh_key:
+        # shlex.quote ensures the key path is safely escaped
+        ssh_parts += ["-i", shlex.quote(ssh_key)]
+    ssh_cmd = " ".join(ssh_parts)
+
+    # user@host:src is passed as a single list element – subprocess does not
+    # invoke a shell, so special characters in the individual arguments do not
+    # enable command injection via shell word splitting.
+    cmd = ["rsync", "-avz", "--progress", f"-e={ssh_cmd}", f"{user}@{host}:{src}", dst]
+    if extra_args:
+        cmd.extend(extra_args)
+
+    logger.info("Running: %s", " ".join(cmd))
+    subprocess.run(cmd, check=True)
+    logger.info("rsync transfer completed: %s@%s:%s -> %s", user, host, src, dst)

autocopy_tool/modules/local_transfer.py

@@ -0,0 +1,40 @@
+"""Structured logging utility for autocopy_tool."""
+import logging
+import sys
+from pathlib import Path
+
+
+def get_logger(name: str, log_file: str = None, level: int = logging.INFO) -> logging.Logger:
+    """Return a configured logger.
+
+    Args:
+        name: Logger name (typically ``__name__`` of the calling module).
+        log_file: Optional path to write log output to a file.
+        level: Logging level (default: INFO).
+
+    Returns:
+        Configured :class:`logging.Logger` instance.
+    """
+    logger = logging.getLogger(name)
+    if logger.handlers:
+        return logger
+
+    logger.setLevel(level)
+    formatter = logging.Formatter(
+        fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    # Console handler
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setFormatter(formatter)
+    logger.addHandler(console_handler)
+
+    # Optional file handler
+    if log_file:
+        Path(log_file).parent.mkdir(parents=True, exist_ok=True)
+        file_handler = logging.FileHandler(log_file, encoding="utf-8")
+        file_handler.setFormatter(formatter)
+        logger.addHandler(file_handler)
+
+    return logger

autocopy_tool/modules/rsync_transfer.py

@@ -0,0 +1,28 @@
+"""Date and time utilities for autocopy_tool."""
+import datetime
+
+
+def today() -> str:
+    """Return today's date as a ``YYYY-MM-DD`` string."""
+    return datetime.date.today().isoformat()
+
+
+def validate_date(date_str: str) -> str:
+    """Validate and normalise a date string to ``YYYY-MM-DD`` format.
+
+    Args:
+        date_str: Date string to validate (e.g. ``"2025-11-04"``).
+
+    Returns:
+        The validated date string in ``YYYY-MM-DD`` format.
+
+    Raises:
+        ValueError: If *date_str* cannot be parsed as a valid date.
+    """
+    try:
+        parsed = datetime.date.fromisoformat(date_str)
+    except ValueError as exc:
+        raise ValueError(
+            f"Invalid date '{date_str}'. Expected format: YYYY-MM-DD"
+        ) from exc
+    return parsed.isoformat()